diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6fbffed47..502e70756 100644 --- a/.github/workflows/ci.yml +++ b/.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 '' - 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 diff --git a/cmd/pad/cmd_materialize.go b/cmd/pad/cmd_materialize.go index a9276a969..a7ca704be 100644 --- a/cmd/pad/cmd_materialize.go +++ b/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) }, } diff --git a/docs/deployment.md b/docs/deployment.md index 4fa44e618..1a8001254 100644 --- a/docs/deployment.md +++ b/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. diff --git a/internal/materialize/adversarial_test.go b/internal/materialize/adversarial_test.go new file mode 100644 index 000000000..45e1efa73 --- /dev/null +++ b/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) + } +} diff --git a/internal/materialize/inflate_darwin_test.go b/internal/materialize/inflate_darwin_test.go new file mode 100644 index 000000000..8d4d7c13f --- /dev/null +++ b/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 diff --git a/internal/materialize/inflate_linux_test.go b/internal/materialize/inflate_linux_test.go new file mode 100644 index 000000000..dc7841c29 --- /dev/null +++ b/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 diff --git a/internal/materialize/inflate_other_test.go b/internal/materialize/inflate_other_test.go new file mode 100644 index 000000000..70b19b300 --- /dev/null +++ b/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 diff --git a/internal/materialize/inflate_windows_test.go b/internal/materialize/inflate_windows_test.go new file mode 100644 index 000000000..5d819647a --- /dev/null +++ b/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 diff --git a/internal/materialize/measure_linux_test.go b/internal/materialize/measure_linux_test.go new file mode 100644 index 000000000..0fa44667b --- /dev/null +++ b/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//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//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() + } +} diff --git a/internal/materialize/memcap_darwin.go b/internal/materialize/memcap_darwin.go new file mode 100644 index 000000000..9a999ff40 --- /dev/null +++ b/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 ` (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) +} diff --git a/internal/materialize/memcap_linux.go b/internal/materialize/memcap_linux.go new file mode 100644 index 000000000..f1df78f2e --- /dev/null +++ b/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//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//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)) +} diff --git a/internal/materialize/memcap_other.go b/internal/materialize/memcap_other.go new file mode 100644 index 000000000..73a86feeb --- /dev/null +++ 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} diff --git a/internal/materialize/memcap_ps.go b/internal/materialize/memcap_ps.go new file mode 100644 index 000000000..9ec26accb --- /dev/null +++ b/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 `, 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 +} diff --git a/internal/materialize/memcap_ps_test.go b/internal/materialize/memcap_ps_test.go new file mode 100644 index 000000000..e87490701 --- /dev/null +++ b/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) + } + } +} diff --git a/internal/materialize/memcap_statparse.go b/internal/materialize/memcap_statparse.go new file mode 100644 index 000000000..ec5fe1b19 --- /dev/null +++ b/internal/materialize/memcap_statparse.go @@ -0,0 +1,22 @@ +package materialize + +import ( + "fmt" + "strconv" + "strings" +) + +// parseStatVsize extracts vsize (field 23) from a /proc//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) +} diff --git a/internal/materialize/memcap_windows.go b/internal/materialize/memcap_windows.go new file mode 100644 index 000000000..4b7601fff --- /dev/null +++ b/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() + } + } + } +} diff --git a/internal/materialize/orphan.go b/internal/materialize/orphan.go new file mode 100644 index 000000000..82fbecc95 --- /dev/null +++ b/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) +} diff --git a/internal/materialize/orphan_linux.go b/internal/materialize/orphan_linux.go new file mode 100644 index 000000000..d71885e64 --- /dev/null +++ b/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 +} diff --git a/internal/materialize/orphan_nolinux.go b/internal/materialize/orphan_nolinux.go new file mode 100644 index 000000000..5dc3a039c --- /dev/null +++ b/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) {} diff --git a/internal/materialize/orphanwatch_other.go b/internal/materialize/orphanwatch_other.go new file mode 100644 index 000000000..402f01b55 --- /dev/null +++ b/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() {} diff --git a/internal/materialize/orphanwatch_unix.go b/internal/materialize/orphanwatch_unix.go new file mode 100644 index 000000000..1de86819f --- /dev/null +++ b/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() + } + }() +} diff --git a/internal/materialize/procgone_other_test.go b/internal/materialize/procgone_other_test.go new file mode 100644 index 000000000..4ba593650 --- /dev/null +++ b/internal/materialize/procgone_other_test.go @@ -0,0 +1,5 @@ +//go:build !unix && !windows + +package materialize + +func processGone(pid int) error { return nil } diff --git a/internal/materialize/procgone_unix_test.go b/internal/materialize/procgone_unix_test.go new file mode 100644 index 000000000..fc60c534b --- /dev/null +++ b/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) +} diff --git a/internal/materialize/procgone_windows_test.go b/internal/materialize/procgone_windows_test.go new file mode 100644 index 000000000..99ec28b3b --- /dev/null +++ b/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 +} diff --git a/internal/materialize/race_off_test.go b/internal/materialize/race_off_test.go new file mode 100644 index 000000000..cfd991a06 --- /dev/null +++ b/internal/materialize/race_off_test.go @@ -0,0 +1,5 @@ +//go:build !race + +package materialize + +const raceEnabled = false diff --git a/internal/materialize/race_on_test.go b/internal/materialize/race_on_test.go new file mode 100644 index 000000000..59879bc76 --- /dev/null +++ b/internal/materialize/race_on_test.go @@ -0,0 +1,5 @@ +//go:build race + +package materialize + +const raceEnabled = true diff --git a/internal/materialize/refuse_test.go b/internal/materialize/refuse_test.go new file mode 100644 index 000000000..9b3d282eb --- /dev/null +++ b/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) + } + } +} diff --git a/internal/materialize/rss_darwin_test.go b/internal/materialize/rss_darwin_test.go new file mode 100644 index 000000000..6ab3a79ab --- /dev/null +++ b/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 diff --git a/internal/materialize/rss_linux_test.go b/internal/materialize/rss_linux_test.go new file mode 100644 index 000000000..e5908c120 --- /dev/null +++ b/internal/materialize/rss_linux_test.go @@ -0,0 +1,26 @@ +package materialize + +import ( + "fmt" + "os" + "strconv" + "strings" +) + +// testRSSSampler reads /proc//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 +} diff --git a/internal/materialize/rss_other_test.go b/internal/materialize/rss_other_test.go new file mode 100644 index 000000000..257648552 --- /dev/null +++ b/internal/materialize/rss_other_test.go @@ -0,0 +1,5 @@ +//go:build !linux && !darwin && !windows + +package materialize + +var testRSSSampler func(pid int) (uint64, error) diff --git a/internal/materialize/rss_windows_test.go b/internal/materialize/rss_windows_test.go new file mode 100644 index 000000000..1402e3236 --- /dev/null +++ b/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 +} diff --git a/internal/materialize/skip_test.go b/internal/materialize/skip_test.go new file mode 100644 index 000000000..8f631b9c2 --- /dev/null +++ b/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) +} diff --git a/internal/materialize/supervisor.go b/internal/materialize/supervisor.go new file mode 100644 index 000000000..e7b9bbc2d --- /dev/null +++ b/internal/materialize/supervisor.go @@ -0,0 +1,1096 @@ +package materialize + +import ( + "bytes" + "context" + "encoding/base64" + "encoding/binary" + "encoding/json" + "errors" + "fmt" + "io" + "log/slog" + "os" + "os/exec" + "runtime" + "strings" + "sync" + "sync/atomic" + "time" +) + +// The Supervisor runs materializer jobs in a child process +// (`pad __materialize-worker`, RunWorker) and never in the server's own heap. +// +// Why a child: a crafted ~1 MB Yjs update drove an in-process runner past +// 5 GB, into Go's fatal, uncatchable out-of-memory — it would have taken the +// whole server down. In a child that is one dead worker. Why ONE long-lived +// child: loading the bundle costs ~2.6 s, so a process per job would spend +// more time starting than working. Why KILL on the deadline: goja's Interrupt +// cannot preempt a single long native call, so the soft deadline inside the +// child (timeout_ms) is not a bound; the process kill is. +// +// It is not wired to anything yet (TASK-2198 U4 does that). Callers are +// background workers: Materialize blocks for as long as the job, a respawn +// backoff, or waiting behind another job takes, bounded only by ctx. + +var ( + // ErrDeadline: the job ran past its deadline. When the child's own + // interrupt stopped it the child survives; otherwise the child was killed. + ErrDeadline = errors.New("materialize: job deadline exceeded") + // ErrChildDied: the worker process died, or was killed, while it held + // the job. The job is NOT retried here: the input may be what killed it, + // and retry budgets on untrusted input belong to the caller. + ErrChildDied = errors.New("materialize: worker process died") + // ErrMemoryLimit: the worker died on its memory cap. It wraps + // ErrChildDied as well. Detected from the Go runtime's out-of-memory + // report on the child's stderr (Linux, Windows) or from the RSS + // watchdog's own kill (macOS); a death it cannot attribute is plain + // ErrChildDied. + ErrMemoryLimit = errors.New("materialize: worker exceeded its memory limit") + // ErrProtocol: the worker answered with something that is not a valid + // response to the request. The worker is killed. + ErrProtocol = errors.New("materialize: worker protocol error") + // ErrStart: the worker could not be started, capped or made ready. + ErrStart = errors.New("materialize: worker failed to start") + // ErrJobFailed: the worker answered the job with an error (a JavaScript + // exception, a malformed row set). The worker is healthy. + ErrJobFailed = errors.New("materialize: job failed") + // ErrClosed: the Supervisor is closed. + ErrClosed = errors.New("materialize: supervisor closed") + // ErrJobTooLarge: the encoded request exceeds MaxFrameBytes. Nothing was + // sent. + ErrJobTooLarge = errors.New("materialize: job exceeds the worker frame limit") + // ErrNoMemoryCap: no memory cap could be established for a worker — the + // OS has no cap mechanism, or the baseline probe failed or read an + // implausible value — so the job was refused and NOTHING was sent to a + // worker. Materialization is a recovery enhancement; without a cap it is + // off, never uncapped. + ErrNoMemoryCap = errors.New("materialize: no memory cap could be established; job refused") +) + +// MaxPlausibleBaseline bounds the memory baseline a loaded worker may report +// before the cap is armed. The largest measured (Linux address space, real +// worker, TestSupervisorBaselineMeasurements) is 1704 MiB at GOMAXPROCS=64; +// resident and committed baselines are far smaller (~75 MiB RSS). 8 GiB is +// ~4.8x the largest, so a reading above it is not a loaded-but-idle worker: +// it is a broken probe, or a worker that ran away before its first job, and +// a cap derived from it would be meaningless. +const MaxPlausibleBaseline = 8 << 30 + +// workerGOMAXPROCS is the worker's GOMAXPROCS. It runs one job at a time and +// gains nothing from more Ps; fewer Ps means fewer threads and a smaller, +// steadier address-space baseline. +const workerGOMAXPROCS = "2" + +// WorkerCommand is the argv[1] that runs the worker (cmd/pad/cmd_materialize.go). +const WorkerCommand = "__materialize-worker" + +// SupervisorConfig configures a Supervisor. The zero value is usable. +type SupervisorConfig struct { + // Timeout is the per-job hard deadline (0: DefaultTimeout), clamped to + // [MinTimeout, MaxTimeout]. The child is asked to stop itself a little + // earlier (softTimeout). + Timeout time.Duration + // MemLimit is the worker's memory cap in bytes (0: DefaultMemLimit), + // clamped to [MinMemLimit, MaxMemLimit]. What it measures is per OS; see + // memCapMechanism. + MemLimit uint64 + // IdleTimeout stops a worker that has had no job for this long (0: + // DefaultIdleTimeout; NeverIdle: never), clamped to [MinIdleTimeout, + // MaxIdleTimeout]. It is a clean stop, not a death: it is logged at INFO + // and never counts toward the respawn backoff. The next job starts a new + // worker, as the first did. + IdleTimeout time.Duration + // Command builds the worker's command. Nil runs this same binary with + // WorkerCommand. Stdin, Stdout and Stderr must be left unset. + Command func() (*exec.Cmd, error) + // Logger receives the start, stop and stderr lines. Nil: slog.Default(). + Logger *slog.Logger + // StartTimeout bounds spawn-to-ready (the bundle load). 0: 60s. + StartTimeout time.Duration + // Respawn backoff after a death: BackoffBase (0: 1s), doubling per + // consecutive death, capped at BackoffMax (0: 5 min). A child that lived + // HealthyStretch (0: 1 min) resets the streak when it dies. + BackoffBase, BackoffMax, HealthyStretch time.Duration + + // Test seams. + now func() time.Time + sleep func(ctx context.Context, d time.Duration) error + capOverride *memCapImpl + watchInterval time.Duration + onSpawn func(pid int) + extraEnv []string // appended last to the worker's environment + idleExact time.Duration // > 0: the idle timeout, unclamped + onIdleStop func() // runs while an idle stop holds the job slot + idleGrace time.Duration // how long an idle stop waits for EOF before a kill (0: 5s) +} + +// memCapImpl is one OS's way of capping the child; memcap_.go defines +// platformMemCap. The cap is RELATIVE: once the worker has loaded its +// bundle, and before it is sent any job, its current usage is read as a +// baseline and the cap is set to baseline + limit. What "usage" is differs +// per OS: address space (Linux), committed memory (Windows), resident set +// (macOS). +type memCapImpl struct { + // mechanism names it in the start log line. + mechanism string + // unsupported: this OS has no cap, so no worker is ever started and every + // job is refused with ErrNoMemoryCap. + unsupported bool + // prepare adjusts the command before Start (nil: nothing). + prepare func(cmd *exec.Cmd) + // attach runs right after Start, before the child is sent anything, and + // returns the handle that measures and caps it. A mechanism that can SEE + // the cap being hit (Windows' job notifications) calls onLimit, which + // kills the child and fails its job with ErrMemoryLimit: the cap must not + // depend on the child choosing to die when an allocation is refused. + attach func(cmd *exec.Cmd, onLimit func()) (*capHandle, error) + // rss, instead of attach: the parent reads the child's resident set as + // the baseline and then polls it every watchInterval, killing the child + // above baseline + limit. + rss func(pid int) (uint64, error) +} + +// capHandle measures and caps one started child. +type capHandle struct { + baseline func() (uint64, error) + set func(capBytes uint64) error + release func() // after the child is reaped; may be nil +} + +// Supervisor owns at most one worker process and runs jobs through it one at +// a time. Safe for concurrent use. +type Supervisor struct { + cfg SupervisorConfig + log *slog.Logger + timeout time.Duration + memLimit uint64 + idle time.Duration // NeverIdle: no idle stop + cap memCapImpl + + slot chan struct{} // held for a whole job: the child is serial + + mu sync.Mutex + closed bool + probeWarned bool // a baseline-probe failure has been logged; cleared by a success + unsupWarned sync.Once + child *child + stopping *child // forgotten by an idle stop that has not reaped it yet + nextID uint64 + streak int // consecutive deaths + nextSpawn time.Time // no spawn before this +} + +// NewSupervisor builds a Supervisor. It starts no process: the first job does. +func NewSupervisor(cfg SupervisorConfig) *Supervisor { + if cfg.Logger == nil { + cfg.Logger = slog.Default() + } + if cfg.Command == nil { + cfg.Command = defaultWorkerCommand + } + if cfg.StartTimeout <= 0 { + cfg.StartTimeout = 60 * time.Second + } + if cfg.BackoffBase <= 0 { + cfg.BackoffBase = time.Second + } + if cfg.BackoffMax <= 0 { + cfg.BackoffMax = 5 * time.Minute + } + if cfg.HealthyStretch <= 0 { + cfg.HealthyStretch = time.Minute + } + if cfg.now == nil { + cfg.now = time.Now + } + if cfg.sleep == nil { + cfg.sleep = sleepCtx + } + if cfg.watchInterval <= 0 { + cfg.watchInterval = 100 * time.Millisecond + } + s := &Supervisor{cfg: cfg, log: cfg.Logger, slot: make(chan struct{}, 1), cap: platformMemCap} + if cfg.capOverride != nil { + s.cap = *cfg.capOverride + } + var clamped bool + if s.timeout, clamped = effectiveTimeout(cfg.Timeout); clamped { + s.log.Warn("materialize: "+EnvTimeout+" out of range; clamped", + "requested", cfg.Timeout.String(), "effective", s.timeout.String(), + "min", MinTimeout.String(), "max", MaxTimeout.String()) + } + if s.memLimit, clamped = effectiveMemLimit(cfg.MemLimit); clamped { + s.log.Warn("materialize: "+EnvMemLimit+" out of range; clamped", + "requested", formatBytes(cfg.MemLimit), "effective", formatBytes(s.memLimit), + "min", formatBytes(MinMemLimit), "max", formatBytes(MaxMemLimit)) + } + if cfg.idleExact > 0 { + s.idle = cfg.idleExact + } else if s.idle, clamped = effectiveIdleTimeout(cfg.IdleTimeout); clamped { + s.log.Warn("materialize: "+EnvIdleTimeout+" out of range; clamped", + "requested", cfg.IdleTimeout.String(), "effective", s.idle.String(), + "min", MinIdleTimeout.String(), "max", MaxIdleTimeout.String()) + } + return s +} + +func defaultWorkerCommand() (*exec.Cmd, error) { + exe, err := os.Executable() + if err != nil { + return nil, err + } + return exec.Command(exe, WorkerCommand), nil +} + +func sleepCtx(ctx context.Context, d time.Duration) error { + t := time.NewTimer(d) + defer t.Stop() + select { + case <-t.C: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + +// softTimeout is the deadline the child enforces on itself with goja's +// Interrupt: a tenth short of the hard one (at least 50 ms short), so a job +// the interrupt CAN stop fails without costing a respawn. +func softTimeout(hard time.Duration) time.Duration { + margin := hard / 10 + if margin < 50*time.Millisecond { + margin = 50 * time.Millisecond + } + return hard - margin +} + +// Materialize runs job in the worker and returns its markdown. It starts the +// worker if none is running (waiting out any respawn backoff first), and +// waits behind a job already running. The job is never retried: an error +// from a dead or killed worker is returned as is (see the Err values). +// +// When ctx ends while the job runs, the worker is killed (it cannot be told +// to drop a job) and ctx's error is returned, wrapped. +func (s *Supervisor) Materialize(ctx context.Context, job Job) (string, error) { + select { + case s.slot <- struct{}{}: + case <-ctx.Done(): + return "", ctx.Err() + } + defer func() { <-s.slot }() + + if s.cap.unsupported { + s.unsupWarned.Do(func() { + s.log.Warn("materialize: no memory cap mechanism on this OS; op-log materialization is disabled", + "goos", runtime.GOOS) + }) + return "", fmt.Errorf("%w: no cap mechanism on %s", ErrNoMemoryCap, runtime.GOOS) + } + c, err := s.ensureChild(ctx) + if err != nil { + return "", err + } + md, err := s.run(ctx, c, job) + s.armIdle(c) + return md, err +} + +// armIdle (re)starts c's idle timer after a job, if c is still the worker (a +// job that killed it has already forgotten it). Called holding the job slot. +func (s *Supervisor) armIdle(c *child) { + if s.idle <= 0 { + return + } + s.mu.Lock() + defer s.mu.Unlock() + if s.child != c || s.closed { + return + } + c.lastJob = time.Now() + if c.idleTimer == nil { + c.idleTimer = time.AfterFunc(s.idle, func() { s.idleStop(c) }) + } else { + c.idleTimer.Reset(s.idle) + } +} + +// idleStop ends a worker that has had no job for s.idle. It holds the job +// slot for the whole stop, so no job can be sent to a worker that is going; +// a job that arrives meanwhile waits for the slot, finds no worker, and starts +// a new one (ensureChild), with no backoff, because an idle stop records no +// death. If the slot is taken, a job is running, and its end re-arms the timer. +func (s *Supervisor) idleStop(c *child) { + select { + case s.slot <- struct{}{}: + default: + return + } + defer func() { <-s.slot }() + s.mu.Lock() + if s.child != c || s.closed { + s.mu.Unlock() + return + } + // A job that ended just as the timer fired re-armed it: not idle yet. + if left := s.idle - time.Since(c.lastJob); left > 0 { + c.idleTimer.Reset(left) + s.mu.Unlock() + return + } + s.child = nil // forgotten, NOT recorded as a death + s.stopping = c // Close still ends and reaps it + s.mu.Unlock() + defer func() { + s.mu.Lock() + if s.stopping == c { + s.stopping = nil + } + s.mu.Unlock() + }() + if s.cfg.onIdleStop != nil { + s.cfg.onIdleStop() + } + // A clean end: close its stdin, and the worker's read loop ends on EOF + // with status 0. Kill it only if it does not go. + // Claims the reason only if nothing has: Close may already have killed it. + c.killReason.CompareAndSwap(nil, "idle") + _ = c.stdin.Close() + grace := s.cfg.idleGrace + if grace <= 0 { + grace = 5 * time.Second + } + if !c.awaitExitFor(grace) { + c.kill("idle") + c.awaitExit() + } + // Close may have ended it meanwhile (s.stopping); say so rather than idle. + reason := "idle" + if c.killedFor() == "closed" { + reason = "closed" + } + s.log.Info("materialize worker stopped", + "pid", c.pid, "reason", reason, "exit", c.exitString(), + "idle_for", s.idle.String(), + "uptime", s.cfg.now().Sub(c.started).Round(time.Millisecond).String(), "jobs", c.jobs.Load()) +} + +// Close kills the worker, if any, and refuses every later job with ErrClosed. +// Idempotent. A job running when Close is called fails with ErrClosed. +func (s *Supervisor) Close() error { + s.mu.Lock() + if s.closed { + s.mu.Unlock() + return nil + } + s.closed = true + c := s.child + s.child = nil + if c != nil && c.idleTimer != nil { + c.idleTimer.Stop() + } + // A worker an idle stop is still ending: Close does not return while it + // runs (codex). The idle stop logs its own stop line. + stopping := s.stopping + s.mu.Unlock() + if stopping != nil { + stopping.kill("closed") + stopping.awaitExit() + } + if c != nil { + c.kill("closed") + c.awaitExit() + s.logStop(c, "closed", 0) + } + return nil +} + +// --------------------------------------------------------------------------- +// one worker process + +type child struct { + cmd *exec.Cmd + pid int + stdin *os.File + started time.Time + jobs atomic.Int64 + + frames chan []byte // responses, in order + readErr error // set before readEnd closes + readEnd chan struct{} // reader stopped (EOF, error, or quit) + exited chan struct{} // reaped; waitErr set + waitErr error + quit chan struct{} // closed by kill + + killOnce sync.Once + killReason atomic.Value // string: why WE killed it + stderr *stderrLog + + idleTimer *time.Timer // guarded by Supervisor.mu + lastJob time.Time // guarded by Supervisor.mu; wall clock, as the timer is + + gomaxprocs string + baseline, capBytes uint64 // set by armCap before any job + rssCap atomic.Uint64 // the watchdog's threshold (macOS) +} + +func (c *child) kill(reason string) { + c.killOnce.Do(func() { + c.killReason.Store(reason) + close(c.quit) + _ = c.cmd.Process.Kill() + }) +} + +func (c *child) killedFor() string { + r, _ := c.killReason.Load().(string) + return r +} + +// awaitExit waits for the child to be reaped. Kill is SIGKILL / +// TerminateProcess, so this is prompt; the bound covers a wedged kernel. +func (c *child) awaitExit() bool { + return c.awaitExitFor(10 * time.Second) +} + +func (c *child) awaitExitFor(d time.Duration) bool { + select { + case <-c.exited: + return true + case <-time.After(d): + return false + } +} + +// exitString describes how the child ended, once it has been reaped. +func (c *child) exitString() string { + select { + case <-c.exited: + return exitDetail(c.waitErr) + default: + return "not reaped" + } +} + +func (s *Supervisor) ensureChild(ctx context.Context) (*child, error) { + s.mu.Lock() + if s.closed { + s.mu.Unlock() + return nil, ErrClosed + } + if c := s.child; c != nil { + s.mu.Unlock() + return c, nil + } + wait := s.nextSpawn.Sub(s.cfg.now()) + s.mu.Unlock() + + if wait > 0 { + if err := s.cfg.sleep(ctx, wait); err != nil { + return nil, fmt.Errorf("materialize: waiting out worker respawn backoff: %w", err) + } + } + return s.spawn(ctx) +} + +// spawn starts, caps and readies a worker. +// +// The cap is applied after Start returns and BEFORE the parent writes the +// first byte to the child's stdin (the readiness probe). The child cannot do +// anything with untrusted input before it holds a request, so the only work it +// does uncapped is its own startup; see memcap_.go for each OS's window. +func (s *Supervisor) spawn(ctx context.Context) (*child, error) { + cmd, err := s.cfg.Command() + if err != nil { + return nil, s.startFailed(nil, fmt.Errorf("build command: %w", err)) + } + inR, inW, err := os.Pipe() + if err != nil { + return nil, s.startFailed(nil, err) + } + outR, outW, err := os.Pipe() + if err != nil { + inR.Close() + inW.Close() + return nil, s.startFailed(nil, err) + } + c := &child{ + cmd: cmd, stdin: inW, + frames: make(chan []byte), readEnd: make(chan struct{}), + exited: make(chan struct{}), quit: make(chan struct{}), + } + c.stderr = &stderrLog{log: s.log} + cmd.Stdin = inR + cmd.Stdout = outW + cmd.Stderr = c.stderr + // Wait returns this long after exit even if something still holds the + // stderr pipe open. + cmd.WaitDelay = 2 * time.Second + if s.cap.prepare != nil { + s.cap.prepare(cmd) + } + guardAgainstOrphaning(cmd) // also materialises cmd.Env + cmd.Env = append(cmd.Env, "GOMAXPROCS="+workerGOMAXPROCS) + cmd.Env = append(cmd.Env, s.cfg.extraEnv...) + c.gomaxprocs = lastEnv(cmd.Env, "GOMAXPROCS") + if err := startUnlocked(cmd); err != nil { + inR.Close() + inW.Close() + outR.Close() + outW.Close() + return nil, s.startFailed(nil, err) + } + inR.Close() + outW.Close() + c.pid = cmd.Process.Pid + c.started = s.cfg.now() + c.stderr.setPid(c.pid) // the stderr copier is already running + + var handle *capHandle + var release func() + if s.cap.attach != nil { + handle, err = s.cap.attach(cmd, func() { c.kill("memory") }) + if err != nil { + _ = cmd.Process.Kill() + _ = cmd.Wait() + inW.Close() + outR.Close() + return nil, s.startFailed(c, fmt.Errorf("attach memory cap (%s): %w", s.cap.mechanism, err)) + } + release = handle.release + } + + go func() { + c.waitErr = cmd.Wait() + if release != nil { + release() + } + c.stderr.flush() + inW.Close() + close(c.exited) + }() + go func() { + defer close(c.readEnd) + defer outR.Close() + for { + p, err := readFrame(outR) + if err != nil { + c.readErr = err + return + } + select { + case c.frames <- p: + case <-c.quit: + return + } + } + }() + s.mu.Lock() + if s.closed { + s.mu.Unlock() + c.kill("closed") + c.awaitExit() + return nil, ErrClosed + } + s.child = c + s.mu.Unlock() + if s.cfg.onSpawn != nil { + s.cfg.onSpawn(c.pid) + } + + // Readiness: the worker reads nothing until its bundle is loaded, so the + // first answer marks the end of the load. The probe carries an empty + // schema version, which the worker refuses (ErrSchemaVersion) without + // doing any work; the refusal IS the ready signal. Doing this apart from + // the first job keeps the ~2.6 s load out of that job's deadline. + id := s.newID() + probe, err := encodeFrame(WorkerRequest{ID: id}) + if err != nil { + return nil, s.fail(c, "start", fmt.Errorf("%w: encode probe: %v", ErrStart, err)) + } + timer := time.NewTimer(s.cfg.StartTimeout) + defer timer.Stop() + resp, err := s.exchange(ctx, c, probe, timer.C, "start", + fmt.Errorf("%w: not ready within %s; worker killed", ErrStart, s.cfg.StartTimeout)) + if err != nil { + if errors.Is(err, ErrChildDied) && !errors.Is(err, ErrMemoryLimit) { + err = fmt.Errorf("%w: %w", ErrStart, err) + } + return nil, err + } + if resp.ID != id || resp.Error == "" { + return nil, s.fail(c, "protocol", fmt.Errorf("%w: unexpected answer to the readiness probe: %+v", ErrProtocol, resp)) + } + + // The worker is loaded and has been sent nothing but the probe: measure + // it and cap it before the first job. No job ever goes to a worker whose + // cap is not confirmed. + if err := s.armCap(c, handle); err != nil { + return nil, err + } + + s.log.Info("materialize worker started", + "pid", c.pid, + "ready_ms", s.cfg.now().Sub(c.started).Milliseconds(), + "timeout", s.timeout.String(), + "soft_timeout", softTimeout(s.timeout).String(), + "mem_cap", s.cap.mechanism, + "mem_baseline", formatBytes(c.baseline), + "mem_limit", formatBytes(s.memLimit), + "mem_cap_effective", formatBytes(c.capBytes), + "mem_baseline_bytes", c.baseline, + "mem_limit_bytes", s.memLimit, + "mem_cap_effective_bytes", c.capBytes, + "gomaxprocs", c.gomaxprocs) + return c, nil +} + +// armCap reads the loaded worker's baseline and caps it at baseline + +// memLimit. A baseline that cannot be read, or is implausible, goes to +// onBaselineProbeFailure; a cap that cannot be set fails the start. +func (s *Supervisor) armCap(c *child, h *capHandle) error { + var baseline func() (uint64, error) + var set func(uint64) error + switch { + case s.cap.rss != nil: + baseline = func() (uint64, error) { return s.cap.rss(c.pid) } + set = func(capBytes uint64) error { + c.rssCap.Store(capBytes) + go s.watchRSS(c) + return nil + } + case h != nil: + baseline, set = h.baseline, h.set + default: + return s.onBaselineProbeFailure(c, errors.New("no cap mechanism attached"), 0) + } + b, err := baseline() + if err == nil && (b == 0 || b > MaxPlausibleBaseline) { + return s.onBaselineProbeFailure(c, fmt.Errorf("implausible baseline %d bytes (plausible: 1..%d)", b, uint64(MaxPlausibleBaseline)), b) + } + if err != nil { + return s.onBaselineProbeFailure(c, err, 0) + } + capBytes := b + s.memLimit + if err := set(capBytes); err != nil { + return s.fail(c, "start", fmt.Errorf("%w: set memory cap (%s) to %d: %v", ErrStart, s.cap.mechanism, capBytes, err)) + } + c.baseline, c.capBytes = b, capBytes + s.mu.Lock() + s.probeWarned = false // a later failure is news again + s.mu.Unlock() + return nil +} + +// onBaselineProbeFailure is the one place that decides what happens when a +// worker's memory baseline cannot be established (lead ruling: REFUSE). The +// worker is killed without being sent a job, and the job fails with +// ErrNoMemoryCap. One WARN names the cause; further failures stay silent +// until a probe succeeds again, because every refused job starts (and +// probes) a new worker and a WARN per job would be per-job noise — the +// caller counts refusals against its own failure budget. +func (s *Supervisor) onBaselineProbeFailure(c *child, cause error, value uint64) error { + err := fmt.Errorf("%w: %s baseline probe failed: %v", ErrNoMemoryCap, s.cap.mechanism, cause) + s.mu.Lock() + first := !s.probeWarned + s.probeWarned = true + s.mu.Unlock() + if first { + s.log.Warn("materialize: could not establish the worker's memory cap; refusing to materialize until a worker's probe succeeds", + "pid", c.pid, "mechanism", s.cap.mechanism, "cause", cause.Error(), "value", value) + } + c.kill("no_memory_cap") + c.awaitExit() + s.stoppedQuiet(c) + return err +} + +// lastEnv is the value os/exec gives key in env: the last one wins. +func lastEnv(env []string, key string) string { + v := "" + for _, kv := range env { + if k, val, ok := strings.Cut(kv, "="); ok && k == key { + v = val + } + } + return v +} + +// startUnlocked runs cmd.Start on a fresh goroutine. A new goroutine is never +// locked to its thread, so the fork happens on a thread the runtime will not +// retire, which Pdeathsig depends on (orphan_linux.go). +func startUnlocked(cmd *exec.Cmd) error { + done := make(chan error, 1) + go func() { done <- cmd.Start() }() + return <-done +} + +func (s *Supervisor) newID() uint64 { + s.mu.Lock() + defer s.mu.Unlock() + s.nextID++ + return s.nextID +} + +// startFailed records a failed start as a death (it counts toward backoff). +func (s *Supervisor) startFailed(c *child, err error) error { + err = fmt.Errorf("%w: %v", ErrStart, err) + s.mu.Lock() + delay := s.recordDeathLocked(c) + s.mu.Unlock() + pid := 0 + if c != nil { + pid = c.pid + } + s.log.Warn("materialize worker stopped", "pid", pid, "reason", "start", "detail", err.Error(), + "respawn_after", delay.String()) + return err +} + +// run sends one job and waits for its answer. +func (s *Supervisor) run(ctx context.Context, c *child, job Job) (string, error) { + hard := s.timeout + soft := softTimeout(hard) + // When ctx ends first, have the child stop itself first too, so a job + // its interrupt can reach does not cost the worker. + if dl, ok := ctx.Deadline(); ok { + if rem := dl.Sub(time.Now()); rem < hard { + soft = softTimeout(rem) + } + } + if soft < time.Millisecond { + soft = time.Millisecond + } + req := WorkerRequest{ + ID: s.newID(), + Rows: make([]string, len(job.Rows)), + SchemaVersion: job.SchemaVersion, + LinkIndex: job.LinkIndex, + WorkspaceSlug: job.WorkspaceSlug, + TimeoutMs: soft.Milliseconds(), + } + if req.LinkIndex == nil { + req.LinkIndex = []LinkEntry{} + } + for i, b := range job.Rows { + req.Rows[i] = base64.StdEncoding.EncodeToString(b) + } + frame, err := encodeFrame(req) + if err != nil { + return "", err + } + c.jobs.Add(1) + + timer := time.NewTimer(hard) + defer timer.Stop() + resp, err := s.exchange(ctx, c, frame, timer.C, "deadline", + fmt.Errorf("%w: %s passed; worker killed", ErrDeadline, hard)) + if err != nil { + return "", err + } + if resp.ID != req.ID { + return "", s.fail(c, "protocol", fmt.Errorf("%w: response id %d for request %d", ErrProtocol, resp.ID, req.ID)) + } + switch { + case resp.Error != "": + switch { + case strings.Contains(resp.Error, ErrSchemaVersion.Error()): + return "", fmt.Errorf("%w (worker: %s)", ErrSchemaVersion, resp.Error) + case strings.Contains(resp.Error, ErrInterrupted.Error()): + // The child's own interrupt stopped it: soft deadline, child kept. + return "", fmt.Errorf("%w after %s (stopped by the worker's interrupt): %s", ErrDeadline, soft, resp.Error) + } + return "", fmt.Errorf("%w: %s", ErrJobFailed, resp.Error) + case resp.Markdown == nil: + return "", s.fail(c, "protocol", fmt.Errorf("%w: response %d has neither markdown nor error", ErrProtocol, resp.ID)) + } + return *resp.Markdown, nil +} + +// exchange writes one frame and returns the next response. On the timer it +// kills the child for timerReason; on ctx, for "canceled". Every failure +// that ends the child goes through fail. +func (s *Supervisor) exchange(ctx context.Context, c *child, frame []byte, timer <-chan time.Time, timerReason string, timerErr error) (WorkerResponse, error) { + // The write runs apart from the wait: a child that stops reading would + // otherwise block it for good, and a Windows pipe takes no deadline. A + // kill closes the pipe's far end and ends the write. + wrote := make(chan error, 1) + go func() { + _, err := c.stdin.Write(frame) + wrote <- err + }() + for { + select { + case err := <-wrote: + wrote = nil + if err != nil { + return WorkerResponse{}, s.died(c) + } + case p := <-c.frames: + var resp WorkerResponse + if err := json.Unmarshal(p, &resp); err != nil { + return resp, s.fail(c, "protocol", fmt.Errorf("%w: response is not JSON: %v", ErrProtocol, err)) + } + return resp, nil + case <-c.readEnd: + if c.readErr != nil && !errors.Is(c.readErr, io.EOF) && c.killedFor() == "" { + // A truncated or oversized frame from a live child. + select { + case <-c.exited: + case <-time.After(200 * time.Millisecond): + return WorkerResponse{}, s.fail(c, "protocol", fmt.Errorf("%w: %v", ErrProtocol, c.readErr)) + } + } + return WorkerResponse{}, s.died(c) + case <-c.exited: + return WorkerResponse{}, s.died(c) + case <-timer: + return WorkerResponse{}, s.fail(c, timerReason, timerErr) + case <-ctx.Done(): + return WorkerResponse{}, s.fail(c, "canceled", fmt.Errorf("materialize: %w (worker killed)", ctx.Err())) + } + } +} + +// fail kills the child for reason and returns err. A kill already made for +// another reason (the watchdog, Close) wins: its classification is returned. +func (s *Supervisor) fail(c *child, reason string, err error) error { + c.kill(reason) + if c.killedFor() != reason { + return s.died(c) + } + c.awaitExit() + s.stopped(c, reason, err) + return err +} + +// died classifies a child that is gone (or going) and returns the job's error. +func (s *Supervisor) died(c *child) error { + if c.killedFor() == "" && !c.awaitExitFor(2*time.Second) { + // Its stdout ended (or its stdin broke) but it is still running. + return s.fail(c, "protocol", fmt.Errorf("%w: worker closed its output but did not exit", ErrProtocol)) + } + c.awaitExit() + var reason string + var err error + switch k := c.killedFor(); { + case k == "closed": + reason, err = "closed", ErrClosed + case k == "memory": + reason = "memory" + err = fmt.Errorf("%w (%w): reached its %s cap of %s (baseline %s + limit %s); killed by the supervisor", + ErrMemoryLimit, ErrChildDied, s.cap.mechanism, formatBytes(c.capBytes), formatBytes(c.baseline), formatBytes(s.memLimit)) + case k != "": + // Killed by us for a reason whose own path returns; not reached in + // practice, but never report such a death as a crash. + reason, err = k, fmt.Errorf("%w: killed (%s)", ErrChildDied, k) + case c.stderr.oom.Load(): + reason = "memory" + err = fmt.Errorf("%w (%w): %s; the Go runtime reported out of memory under a %s cap (baseline %s + limit %s)", + ErrMemoryLimit, ErrChildDied, c.exitString(), formatBytes(c.capBytes), formatBytes(c.baseline), formatBytes(s.memLimit)) + default: + reason = "exit" + err = fmt.Errorf("%w: %s", ErrChildDied, c.exitString()) + } + s.stopped(c, reason, err) + return err +} + +func exitDetail(err error) string { + if err == nil { + return "exited with status 0" + } + return err.Error() +} + +// stopped forgets the child, schedules the respawn and logs the stop. Called +// once per child: only the job holding the slot (or Close, which has already +// forgotten it) reaches here. +func (s *Supervisor) stopped(c *child, reason string, err error) { + s.mu.Lock() + if s.child != c { + // Close got there first and logged it. + s.mu.Unlock() + return + } + s.child = nil + var delay time.Duration + if reason != "canceled" && reason != "closed" { + delay = s.recordDeathLocked(c) + } + s.mu.Unlock() + s.logStop(c, reason, delay, "detail", err.Error()) +} + +// stoppedQuiet forgets a child killed by onBaselineProbeFailure and schedules +// the respawn, like stopped, without the stop line (that path logs its own +// single WARN). +func (s *Supervisor) stoppedQuiet(c *child) { + s.mu.Lock() + defer s.mu.Unlock() + if s.child != c { + return + } + s.child = nil + s.recordDeathLocked(c) +} + +func (s *Supervisor) logStop(c *child, reason string, delay time.Duration, extra ...any) { + attrs := []any{"pid", c.pid, "reason", reason, "exit", c.exitString(), + "uptime", s.cfg.now().Sub(c.started).Round(time.Millisecond).String(), "jobs", c.jobs.Load(), + "respawn_after", delay.String()} + s.log.Warn("materialize worker stopped", append(attrs, extra...)...) +} + +// recordDeathLocked bumps the death streak and sets the next spawn time. A +// child that lived HealthyStretch resets the streak first. Holds s.mu. +func (s *Supervisor) recordDeathLocked(c *child) time.Duration { + now := s.cfg.now() + if c != nil && now.Sub(c.started) >= s.cfg.HealthyStretch { + s.streak = 0 + } + s.streak++ + delay := s.cfg.BackoffBase + for i := 1; i < s.streak && delay < s.cfg.BackoffMax; i++ { + delay *= 2 + } + if delay > s.cfg.BackoffMax { + delay = s.cfg.BackoffMax + } + s.nextSpawn = now.Add(delay) + return delay +} + +// watchRSS is the parent-side memory cap for platforms that do not enforce +// one on the child (macOS). +func (s *Supervisor) watchRSS(c *child) { + t := time.NewTicker(s.cfg.watchInterval) + defer t.Stop() + for { + select { + case <-c.exited: + return + case <-t.C: + rss, err := s.cap.rss(c.pid) + if err != nil { + continue // gone, or not yet visible; exit is seen above + } + if rss > c.rssCap.Load() { + c.kill("memory") + return + } + } + } +} + +func encodeFrame(req WorkerRequest) ([]byte, error) { + body, err := json.Marshal(req) + if err != nil { + return nil, fmt.Errorf("materialize: encode request: %w", err) + } + if len(body) > MaxFrameBytes { + return nil, fmt.Errorf("%w: %d bytes", ErrJobTooLarge, len(body)) + } + out := make([]byte, 4, 4+len(body)) + binary.BigEndian.PutUint32(out, uint32(len(body))) + return append(out, body...), nil +} + +// --------------------------------------------------------------------------- +// the child's stderr + +const ( + stderrMaxLine = 4 << 10 + stderrMaxLines = 200 +) + +// stderrLog is the child's stderr: every line goes to the log, prefixed by +// the worker's pid, and never near the protocol (stdout). It always drains — +// a child blocked on a full stderr pipe would read as hung — but logs at most +// stderrMaxLines lines per child (a Go crash dump is thousands), each cut at +// stderrMaxLine bytes, and reports how many it dropped. +// +// It also watches for the Go runtime's out-of-memory report, which is how a +// death on the RLIMIT_AS / Job Object cap is told from any other. +type stderrLog struct { + log *slog.Logger + pid int + + mu sync.Mutex + partial []byte + lines int + dropped int + oom atomic.Bool +} + +// oomMarkers are the Go runtime's words for a failed allocation: "fatal +// error: runtime: out of memory" / "fatal error: out of memory", and the mmap +// failure ("cannot allocate memory", ENOMEM). On Windows a VirtualAlloc +// failure is followed by the same "fatal error: out of memory". +// +// A cgo build (`make build`; releases are CGO_ENABLED=0) starts threads with +// pthread_create, whose stack mapping fails under an address-space cap and +// is reported as "pthread_create failed: Resource temporarily unavailable" +// (EAGAIN) followed by SIGABRT — measured at the cap on Linux. +// +// On Windows a commit refused by the Job Object is "VirtualAlloc of N bytes +// failed with errno=1455" (ERROR_COMMITMENT_LIMIT). +var oomMarkers = []string{"out of memory", "cannot allocate memory", "pthread_create failed", "errno=1455"} + +func (l *stderrLog) Write(p []byte) (int, error) { + l.mu.Lock() + defer l.mu.Unlock() + n := len(p) + for len(p) > 0 { + i := bytes.IndexByte(p, '\n') + if i < 0 { + room := stderrMaxLine - len(l.partial) + if room > 0 { + take := min(room, len(p)) + l.partial = append(l.partial, p[:take]...) + } + break + } + room := stderrMaxLine - len(l.partial) + if room > 0 { + l.partial = append(l.partial, p[:min(room, i)]...) + } + l.emitLocked() + p = p[i+1:] + } + return n, nil +} + +func (l *stderrLog) emitLocked() { + line := string(l.partial) + l.partial = l.partial[:0] + for _, m := range oomMarkers { + if strings.Contains(line, m) { + l.oom.Store(true) + } + } + if l.lines >= stderrMaxLines { + l.dropped++ + return + } + l.lines++ + l.log.Info("materialize worker stderr: "+line, "pid", l.pid) +} + +func (l *stderrLog) setPid(pid int) { + l.mu.Lock() + l.pid = pid + l.mu.Unlock() +} + +func (l *stderrLog) flush() { + l.mu.Lock() + defer l.mu.Unlock() + if len(l.partial) > 0 { + l.emitLocked() + } + if l.dropped > 0 { + l.log.Info("materialize worker stderr: lines dropped", "pid", l.pid, "dropped", l.dropped) + } +} diff --git a/internal/materialize/supervisor_config.go b/internal/materialize/supervisor_config.go new file mode 100644 index 000000000..51e39b039 --- /dev/null +++ b/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) +} diff --git a/internal/materialize/supervisor_test.go b/internal/materialize/supervisor_test.go new file mode 100644 index 000000000..dc4f5e2e5 --- /dev/null +++ b/internal/materialize/supervisor_test.go @@ -0,0 +1,1156 @@ +package materialize + +import ( + "bufio" + "context" + "encoding/binary" + "encoding/json" + "errors" + "fmt" + "log/slog" + "os" + "os/exec" + "strconv" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + pad "github.com/PerpetualSoftware/pad" +) + +// The supervisor tests spawn THIS test binary as the worker: TestMain sees +// helperEnv and runs a helper instead of the tests. +// +// - "worker": the real RunWorker over the embedded bundle. +// - "script": answers the readiness probe, then does what each job's +// SchemaVersion says (see runScript), so one child can behave well on one +// job and badly on the next. +const helperEnv = "PAD_MATERIALIZE_TEST_HELPER" + +// inflateEnv makes the script helper inflate its baseline by that many MiB +// before it answers the readiness probe (inflateBaseline). +const inflateEnv = "PAD_MATERIALIZE_TEST_INFLATE_MIB" + +// journalEnv names a file the script helper appends every JOB it receives to +// (the readiness probe excluded), so a test can prove what reached a worker. +const journalEnv = "PAD_MATERIALIZE_TEST_JOURNAL" + +func TestMain(m *testing.M) { + if mode := os.Getenv(helperEnv); mode != "" { + os.Exit(runHelper(mode)) + } + os.Exit(m.Run()) +} + +func runHelper(mode string) int { + if mode == "parent" { + return runHelperParent() + } + // As the real worker command does (cmd/pad/cmd_materialize.go). + ExitWhenOrphaned() + if mode == "worker" { + if err := RunWorker(os.Stdin, os.Stdout, pad.MaterializerJS); err != nil { + fmt.Fprintln(os.Stderr, "helper worker:", err) + return 1 + } + return 0 + } + if mib, _ := strconv.Atoi(os.Getenv(inflateEnv)); mib > 0 { + if err := inflateBaseline(mib); err != nil { + fmt.Fprintln(os.Stderr, "helper: inflate:", err) + return 8 + } + } + warmThreads() // as RunWorker does before readiness + r := bufio.NewReader(os.Stdin) + w := bufio.NewWriter(os.Stdout) + for { + p, err := readFrame(r) + if err != nil { + return 0 + } + var req WorkerRequest + if err := json.Unmarshal(p, &req); err != nil { + return 9 + } + if req.SchemaVersion == "" { // the readiness probe + _ = writeFrame(w, WorkerResponse{ID: req.ID, Error: "materialize: schema version mismatch: probe"}) + continue + } + if j := os.Getenv(journalEnv); j != "" { + f, err := os.OpenFile(j, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o600) + if err == nil { + fmt.Fprintln(f, req.SchemaVersion) + f.Close() + } + } + if !runScript(w, req) { + return 0 + } + } +} + +func runScript(w *bufio.Writer, req WorkerRequest) bool { + ok := func(md string) { _ = writeFrame(w, WorkerResponse{ID: req.ID, Markdown: &md}) } + cmd, arg, _ := strings.Cut(req.SchemaVersion, ":") + switch cmd { + case "echo": + ok("echo:" + arg) + case "timeout": // echoes the soft deadline it was handed + ok(strconv.FormatInt(req.TimeoutMs, 10)) + case "sleep": // answer after arg milliseconds + ms, _ := strconv.Atoi(arg) + time.Sleep(time.Duration(ms) * time.Millisecond) + ok("slept:" + arg) + case "hang": // a native call nothing can interrupt + time.Sleep(time.Hour) + case "exit": + fmt.Fprintln(os.Stderr, "helper: exiting on purpose") + os.Exit(3) + case "alloc": // allocate arg MiB, every byte written with noise + // Noise, not a byte per page: macOS compresses idle pages, and a + // page of zeros and one 1 would compress away out of the resident + // set the watchdog reads. + mib, _ := strconv.Atoi(arg) + b := make([]byte, mib<<20) + x := uint64(0x9E3779B97F4A7C15) + for i := 0; i+8 <= len(b); i += 8 { + x ^= x << 13 + x ^= x >> 7 + x ^= x << 17 + binary.LittleEndian.PutUint64(b[i:], x) + } + ok(strconv.Itoa(len(b))) + case "soft": // what the worker answers when its own interrupt fires + _ = writeFrame(w, WorkerResponse{ID: req.ID, Error: "materialize: interrupted: context deadline exceeded"}) + case "jserror": + _ = writeFrame(w, WorkerResponse{ID: req.ID, Error: "materialize: TypeError: boom"}) + case "badjson": + body := []byte("{nope") + _, _ = w.Write([]byte{0, 0, 0, byte(len(body))}) + _, _ = w.Write(body) + _ = w.Flush() + case "wrongid": + md := "x" + _ = writeFrame(w, WorkerResponse{ID: req.ID + 1000, Markdown: &md}) + case "closeout": // close stdout but keep running + os.Stdout.Close() + time.Sleep(time.Hour) + default: + return false + } + return true +} + +// runHelperParent is a supervising process for TestSupervisorOrphanedWorkerExits: +// it spawns a script worker, prints its pid, then hangs it in a job forever. +func runHelperParent() int { + f := &helperFactory{mode: "script"} + s := NewSupervisor(SupervisorConfig{ + Command: f.cmd, + Timeout: MaxTimeout, + capOverride: testCap(), + Logger: slog.New(slog.NewTextHandler(os.Stderr, nil)), + onSpawn: func(pid int) { fmt.Printf("worker %d\n", pid) }, + }) + if _, err := s.Materialize(context.Background(), script("echo:warm")); err != nil { + fmt.Fprintln(os.Stderr, "helper parent:", err) + return 2 + } + fmt.Println("hanging") + _, err := s.Materialize(context.Background(), script("hang")) + fmt.Fprintln(os.Stderr, "helper parent: hang returned:", err) + return 3 +} + +// --------------------------------------------------------------------------- +// fixtures + +type helperFactory struct { + mode string + calls atomic.Int32 +} + +func (h *helperFactory) cmd() (*exec.Cmd, error) { + h.calls.Add(1) + cmd := exec.Command(os.Args[0], "-test.run=^$") + cmd.Env = append(os.Environ(), helperEnv+"="+h.mode) + return cmd, nil +} + +// logSink records every slog record as "msg k=v k=v". +type logSink struct { + mu sync.Mutex + recs []string +} + +func (l *logSink) Enabled(context.Context, slog.Level) bool { return true } +func (l *logSink) Handle(_ context.Context, r slog.Record) error { + var b strings.Builder + b.WriteString(r.Level.String() + " ") + b.WriteString(r.Message) + r.Attrs(func(a slog.Attr) bool { + fmt.Fprintf(&b, " %s=%v", a.Key, a.Value) + return true + }) + l.mu.Lock() + l.recs = append(l.recs, b.String()) + l.mu.Unlock() + return nil +} +func (l *logSink) WithAttrs([]slog.Attr) slog.Handler { return l } +func (l *logSink) WithGroup(string) slog.Handler { return l } + +// find returns the records containing every one of subs. +func (l *logSink) find(subs ...string) []string { + l.mu.Lock() + defer l.mu.Unlock() + var out []string +next: + for _, r := range l.recs { + for _, s := range subs { + if !strings.Contains(r, s) { + continue next + } + } + out = append(out, r) + } + return out +} + +func (l *logSink) warns() []string { + l.mu.Lock() + defer l.mu.Unlock() + var out []string + for _, r := range l.recs { + if strings.HasPrefix(r, "WARN ") { + out = append(out, r) + } + } + return out +} + +func (l *logSink) dump() string { + l.mu.Lock() + defer l.mu.Unlock() + return strings.Join(l.recs, "\n") +} + +type fakeClock struct { + mu sync.Mutex + t time.Time + sleeps []time.Duration +} + +func newFakeClock() *fakeClock { return &fakeClock{t: time.Unix(1_000_000, 0)} } + +func (f *fakeClock) now() time.Time { + f.mu.Lock() + defer f.mu.Unlock() + return f.t +} + +func (f *fakeClock) advance(d time.Duration) { + f.mu.Lock() + f.t = f.t.Add(d) + f.mu.Unlock() +} + +func (f *fakeClock) sleep(_ context.Context, d time.Duration) error { + f.mu.Lock() + f.sleeps = append(f.sleeps, d) + f.t = f.t.Add(d) + f.mu.Unlock() + return nil +} + +func (f *fakeClock) slept() []time.Duration { + f.mu.Lock() + defer f.mu.Unlock() + return append([]time.Duration(nil), f.sleeps...) +} + +// fakeCap reports a fixed baseline and caps nothing: for tests whose subject +// is not the cap, and for race builds, whose runtime reserves terabytes of +// address space (a baseline no plausibility check accepts). +var fakeCap = &memCapImpl{ + mechanism: "fake(test)", + attach: func(*exec.Cmd, func()) (*capHandle, error) { + return &capHandle{ + baseline: func() (uint64, error) { return 100 << 20, nil }, + set: func(uint64) error { return nil }, + }, nil + }, +} + +// testCap is the cap for tests that are not about the cap: the platform's, +// except under the race detector. +func testCap() *memCapImpl { + if raceEnabled { + return fakeCap + } + return nil +} + +type harness struct { + s *Supervisor + f *helperFactory + log *logSink + clk *fakeClock + pids chan int +} + +func newHarness(t *testing.T, mode string, tweak func(*SupervisorConfig)) *harness { + t.Helper() + h := &harness{f: &helperFactory{mode: mode}, log: &logSink{}, clk: newFakeClock(), pids: make(chan int, 64)} + cfg := SupervisorConfig{ + Command: h.f.cmd, + Logger: slog.New(h.log), + Timeout: 10 * time.Second, + now: h.clk.now, + sleep: h.clk.sleep, + capOverride: testCap(), + onSpawn: func(pid int) { h.pids <- pid }, + } + if tweak != nil { + tweak(&cfg) + } + h.s = NewSupervisor(cfg) + t.Cleanup(func() { + _ = h.s.Close() + if t.Failed() { + t.Logf("supervisor log:\n%s", h.log.dump()) + } + }) + return h +} + +func (h *harness) lastPid(t *testing.T) int { + t.Helper() + select { + case p := <-h.pids: + return p + default: + t.Fatal("no worker was spawned") + return 0 + } +} + +func script(cmd string) Job { return Job{SchemaVersion: cmd} } + +// --------------------------------------------------------------------------- +// 1. lazy start + +func TestSupervisorLazyStart(t *testing.T) { + h := newHarness(t, "script", nil) + time.Sleep(200 * time.Millisecond) + if n := h.f.calls.Load(); n != 0 { + t.Fatalf("NewSupervisor spawned %d processes; want none before the first job", n) + } + h.s.mu.Lock() + c := h.s.child + h.s.mu.Unlock() + if c != nil { + t.Fatal("a child exists before the first job") + } + for i := range 3 { + md, err := h.s.Materialize(context.Background(), script("echo:"+strconv.Itoa(i))) + if err != nil || md != "echo:"+strconv.Itoa(i) { + t.Fatalf("job %d: %q, %v", i, md, err) + } + } + if n := h.f.calls.Load(); n != 1 { + t.Fatalf("%d processes for three jobs; want one long-lived child", n) + } +} + +// --------------------------------------------------------------------------- +// 2. a REAL worker reproduces the in-process Runner + +func TestSupervisorRealWorkerParity(t *testing.T) { + h := newHarness(t, "worker", func(c *SupervisorConfig) { c.Timeout = 30 * time.Second }) + r := runner(t) + cases := loadCorpus(t) + if len(cases) > 4 { + cases = cases[:4] + } + for _, tc := range cases { + job := tc.job(t, r.SchemaVersion()) + want, err := r.Materialize(context.Background(), job) + if err != nil { + t.Fatalf("%s in-process: %v", tc.Name, err) + } + got, err := h.s.Materialize(context.Background(), job) + if err != nil { + t.Fatalf("%s through the supervisor: %v", tc.Name, err) + } + if got != want || got != tc.Expected { + t.Errorf("%s: supervisor %q, in-process %q, corpus %q", tc.Name, got, want, tc.Expected) + } + } + // The real worker's refusal comes back typed. + if _, err := h.s.Materialize(context.Background(), Job{SchemaVersion: "999"}); !errors.Is(err, ErrSchemaVersion) { + t.Errorf("schema mismatch: got %v, want ErrSchemaVersion", err) + } + if n := h.f.calls.Load(); n != 1 { + t.Errorf("%d spawns; want 1", n) + } + if got := h.log.find("materialize worker started", "mem_cap="); len(got) != 1 { + t.Errorf("start lines: %q", got) + } +} + +// --------------------------------------------------------------------------- +// 3. deadline: the child is killed, the next job gets a fresh one + +func TestSupervisorDeadlineKillsChild(t *testing.T) { + h := newHarness(t, "script", func(c *SupervisorConfig) { c.Timeout = MinTimeout }) + if _, err := h.s.Materialize(context.Background(), script("echo:warm")); err != nil { + t.Fatal(err) + } + pid := h.lastPid(t) + + start := time.Now() + _, err := h.s.Materialize(context.Background(), script("hang")) + took := time.Since(start) + if !errors.Is(err, ErrDeadline) { + t.Fatalf("got %v, want ErrDeadline", err) + } + if took > MinTimeout+time.Second { + t.Fatalf("deadline %s enforced after %s", MinTimeout, took) + } + if err := processGone(pid); err != nil { + t.Fatalf("the hung worker survived its deadline: %v", err) + } + if got := h.log.find("materialize worker stopped", "reason=deadline", "pid="+strconv.Itoa(pid)); len(got) != 1 { + t.Fatalf("stop line: %q", got) + } + + md, err := h.s.Materialize(context.Background(), script("echo:after")) + if err != nil || md != "echo:after" { + t.Fatalf("job after the kill: %q, %v", md, err) + } + if pid2 := h.lastPid(t); pid2 == pid { + t.Fatalf("the job after a kill ran on pid %d again", pid) + } + t.Logf("deadline %s enforced in %s", MinTimeout, took.Round(time.Millisecond)) +} + +// The child is asked to stop itself a little before the hard deadline, and a +// job its interrupt stops does not cost the worker. +func TestSupervisorSoftDeadlineKeepsChild(t *testing.T) { + h := newHarness(t, "script", func(c *SupervisorConfig) { c.Timeout = 2 * time.Second }) + md, err := h.s.Materialize(context.Background(), script("timeout")) + if err != nil || md != "1800" { + t.Fatalf("timeout_ms for a 2s deadline: %q, %v; want 1800", md, err) + } + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + md, err = h.s.Materialize(ctx, script("timeout")) + if n, _ := strconv.Atoi(md); err != nil || n <= 0 || n >= 1000 { + t.Fatalf("timeout_ms under a 1s ctx: %q, %v; want below the ctx", md, err) + } + if _, err := h.s.Materialize(context.Background(), script("soft")); !errors.Is(err, ErrDeadline) { + t.Fatalf("got %v, want ErrDeadline", err) + } + if _, err := h.s.Materialize(context.Background(), script("echo:x")); err != nil { + t.Fatal(err) + } + if n := h.f.calls.Load(); n != 1 { + t.Fatalf("a soft deadline respawned the worker (%d spawns)", n) + } +} + +// --------------------------------------------------------------------------- +// 4. crash, and the respawn backoff + +func TestSupervisorCrashAndBackoff(t *testing.T) { + h := newHarness(t, "script", func(c *SupervisorConfig) { + c.BackoffBase = time.Second + c.BackoffMax = 8 * time.Second + c.HealthyStretch = time.Minute + }) + var pids []int + for i := range 6 { + _, err := h.s.Materialize(context.Background(), script("exit")) + if !errors.Is(err, ErrChildDied) || errors.Is(err, ErrMemoryLimit) { + t.Fatalf("crash %d: got %v, want ErrChildDied", i, err) + } + if !strings.Contains(err.Error(), "exit status 3") { + t.Fatalf("crash %d: error does not name the exit status: %v", i, err) + } + pids = append(pids, h.lastPid(t)) + } + // Every death is logged with its reason, and the child's stderr reached + // the log, attributed. + for _, pid := range pids { + p := "pid=" + strconv.Itoa(pid) + if got := h.log.find("materialize worker stopped", "reason=exit", "exit status 3", p); len(got) != 1 { + t.Fatalf("stop line for %s: %q", p, got) + } + if got := h.log.find("materialize worker stderr: helper: exiting on purpose", p); len(got) != 1 { + t.Fatalf("stderr line for %s: %q", p, got) + } + } + want := []time.Duration{1 * time.Second, 2 * time.Second, 4 * time.Second, 8 * time.Second, 8 * time.Second} + if got := h.clk.slept(); fmt.Sprint(got) != fmt.Sprint(want) { + t.Fatalf("respawn waits %v, want %v", got, want) + } + + // The next job respawns (after the capped wait) and succeeds. + if md, err := h.s.Materialize(context.Background(), script("echo:ok")); err != nil || md != "echo:ok" { + t.Fatalf("job after crashes: %q, %v", md, err) + } + // A child that lived a healthy stretch resets the streak: its death + // costs the base wait, not the cap. + h.clk.advance(2 * time.Minute) + if _, err := h.s.Materialize(context.Background(), script("exit")); !errors.Is(err, ErrChildDied) { + t.Fatal(err) + } + if _, err := h.s.Materialize(context.Background(), script("echo:ok")); err != nil { + t.Fatal(err) + } + slept := h.clk.slept() + if got := slept[len(slept)-1]; got != time.Second { + t.Fatalf("wait after a healthy stretch: %v, want 1s (all: %v)", got, slept) + } +} + +func TestSupervisorProtocolErrors(t *testing.T) { + h := newHarness(t, "script", nil) + if _, err := h.s.Materialize(context.Background(), script("echo:warm")); err != nil { + t.Fatal(err) + } + pid := h.lastPid(t) + for _, bad := range []string{"badjson", "wrongid"} { + if _, err := h.s.Materialize(context.Background(), script(bad)); !errors.Is(err, ErrProtocol) { + t.Fatalf("%s: got %v, want ErrProtocol", bad, err) + } + if err := processGone(pid); err != nil { + t.Fatalf("%s: worker not killed: %v", bad, err) + } + if got := h.log.find("materialize worker stopped", "reason=protocol", "pid="+strconv.Itoa(pid)); len(got) != 1 { + t.Fatalf("%s: stop line %q", bad, got) + } + if md, err := h.s.Materialize(context.Background(), script("echo:ok")); err != nil || md != "echo:ok" { + t.Fatalf("after %s: %q, %v", bad, md, err) + } + pid = h.lastPid(t) + } + if _, err := h.s.Materialize(context.Background(), script("jserror")); !errors.Is(err, ErrJobFailed) { + t.Fatalf("jserror: %v", err) + } + if _, err := h.s.Materialize(context.Background(), script("closeout")); !errors.Is(err, ErrProtocol) { + t.Fatalf("closeout: got %v, want ErrProtocol", err) + } +} + +func TestSupervisorCtxCancelKillsWithoutBackoff(t *testing.T) { + h := newHarness(t, "script", nil) + if _, err := h.s.Materialize(context.Background(), script("echo:warm")); err != nil { + t.Fatal(err) + } + pid := h.lastPid(t) + ctx, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond) + defer cancel() + _, err := h.s.Materialize(ctx, script("hang")) + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("got %v", err) + } + if err := processGone(pid); err != nil { + t.Fatal(err) + } + if _, err := h.s.Materialize(context.Background(), script("echo:ok")); err != nil { + t.Fatal(err) + } + if s := h.clk.slept(); len(s) != 0 { + t.Fatalf("a caller's cancellation cost a backoff: %v", s) + } +} + +// --------------------------------------------------------------------------- +// 5. memory cap + +// TestSupervisorMemoryCapPlatform runs the cap this OS ships with, which is +// RELATIVE to the loaded worker: a job may grow it by the limit. It runs +// twice: as is, and with the worker's baseline inflated before readiness +// (inflateBaseline: ~1.5 GiB of never-touched address space on Linux), which +// stands in for a host whose runtime reserves more than this one — the case +// the absolute cap this replaced failed on CI. +func TestSupervisorMemoryCapPlatform(t *testing.T) { + if raceEnabled { + skipCapTest(t, "the race runtime cannot start under an address-space cap") + } + if platformMemCap.unsupported { + skipCapTest(t, "no memory cap on this OS ("+platformMemCap.mechanism+")") + } + const limit = 512 << 20 + for _, inflate := range []int{0, inflateMiB} { + t.Run("baseline+"+strconv.Itoa(inflate)+"MiB", func(t *testing.T) { + h := newHarness(t, "script", func(c *SupervisorConfig) { + c.MemLimit = limit + c.capOverride = nil + c.extraEnv = []string{inflateEnv + "=" + strconv.Itoa(inflate)} + }) + if _, err := h.s.Materialize(context.Background(), script("echo:warm")); err != nil { + t.Fatal(err) + } + pid := h.lastPid(t) + started := h.log.find("materialize worker started", "pid="+strconv.Itoa(pid)) + if len(started) != 1 || !strings.Contains(started[0], "mem_limit=512MiB") || !strings.Contains(started[0], "mem_baseline=") { + t.Fatalf("start line: %q", started) + } + t.Log(started[0]) + if inflate > 0 { + h.s.mu.Lock() + b := h.s.child.baseline + h.s.mu.Unlock() + if b < uint64(inflate)<<20 { + t.Fatalf("baseline %d does not include the %d MiB inflation", b, inflate) + } + } + // Comfortably under baseline + limit: fine. + if md, err := h.s.Materialize(context.Background(), script("alloc:128")); err != nil || md != strconv.Itoa(128<<20) { + t.Fatalf("128 MiB of growth under a 512 MiB limit: %q, %v", md, err) + } + // Over it: killed on the cap. + md, err := h.s.Materialize(context.Background(), script("alloc:1024")) + if !errors.Is(err, ErrMemoryLimit) || !errors.Is(err, ErrChildDied) { + t.Fatalf("1 GiB of growth under a 512 MiB limit: %q, %v; want ErrMemoryLimit", md, err) + } + if got := h.log.find("materialize worker stopped", "reason=memory", "pid="+strconv.Itoa(pid)); len(got) != 1 { + t.Fatalf("stop line: %q", got) + } + if md, err := h.s.Materialize(context.Background(), script("echo:ok")); err != nil || md != "echo:ok" { + t.Fatalf("after the memory kill: %q, %v", md, err) + } + t.Logf("over-cap error: %v", err) + }) + } +} + +// TestSupervisorRSSWatchdog drives the macOS mechanism (a parent-side RSS +// poll and kill) wherever a sampler exists, so it runs on Linux too. +func TestSupervisorRSSWatchdog(t *testing.T) { + if testRSSSampler == nil { + skipCapTest(t, "no RSS sampler for tests on this OS") + } + h := newHarness(t, "script", func(c *SupervisorConfig) { + c.MemLimit = 512 << 20 + c.capOverride = &memCapImpl{mechanism: "rss_watchdog(test)", rss: testRSSSampler} + c.watchInterval = 100 * time.Millisecond + }) + if _, err := h.s.Materialize(context.Background(), script("echo:warm")); err != nil { + t.Fatal(err) + } + pid := h.lastPid(t) + md, err := h.s.Materialize(context.Background(), script("alloc:1024")) + if !errors.Is(err, ErrMemoryLimit) { + t.Fatalf("1 GiB resident under a baseline + 512 MiB watchdog: %q, %v", md, err) + } + if !strings.Contains(err.Error(), "watchdog") { + t.Fatalf("not attributed to the watchdog: %v", err) + } + if err := processGone(pid); err != nil { + t.Fatal(err) + } + if got := h.log.find("materialize worker stopped", "reason=memory"); len(got) != 1 { + t.Fatalf("stop line: %q", got) + } + if _, err := h.s.Materialize(context.Background(), script("echo:ok")); err != nil { + t.Fatal(err) + } +} + +// --------------------------------------------------------------------------- +// 6. env knobs + +func TestParseMemLimit(t *testing.T) { + for in, want := range map[string]uint64{ + "2048MiB": 2 << 30, + "2GiB": 2 << 30, + "1610612736": 1536 << 20, + "512KiB": 512 << 10, + " 3GiB ": 3 << 30, + } { + if got, err := ParseMemLimit(in); err != nil || got != want { + t.Errorf("%q: %d, %v; want %d", in, got, err, want) + } + } + for _, in := range []string{"", "2GB", "2 GiB", "2gib", "1.5GiB", "-1", "GiB", "99999999999999999999", "17179869184GiB"} { + if got, err := ParseMemLimit(in); err == nil { + t.Errorf("%q parsed as %d; want an error", in, got) + } + } +} + +func TestConfigFromEnv(t *testing.T) { + env := func(m map[string]string) func(string) string { return func(k string) string { return m[k] } } + sink := &logSink{} + lg := slog.New(sink) + + cfg := ConfigFromEnv(env(nil), lg) + if cfg.Timeout != 0 || cfg.MemLimit != 0 { + t.Fatalf("unset: %+v", cfg) + } + cfg = ConfigFromEnv(env(map[string]string{EnvTimeout: "5s", EnvMemLimit: "3GiB"}), lg) + if cfg.Timeout != 5*time.Second || cfg.MemLimit != 3<<30 { + t.Fatalf("set: %v %d", cfg.Timeout, cfg.MemLimit) + } + if len(sink.find("materialize")) != 0 { + t.Fatalf("valid values logged: %s", sink.dump()) + } + cfg = ConfigFromEnv(env(map[string]string{EnvTimeout: "soon", EnvMemLimit: "2GB"}), lg) + if cfg.Timeout != 0 || cfg.MemLimit != 0 { + t.Fatalf("invalid values must fall back to the default: %+v", cfg) + } + if len(sink.find("ignoring invalid "+EnvTimeout, "value=soon")) != 1 || len(sink.find("ignoring invalid "+EnvMemLimit, "value=2GB")) != 1 { + t.Fatalf("invalid values not logged: %s", sink.dump()) + } +} + +func TestSupervisorClampAndEffectiveLog(t *testing.T) { + cases := []struct { + timeout time.Duration + mem uint64 + wantT string + wantM string + clampLogs int + }{ + {0, 0, "2s", "2GiB", 0}, + {time.Millisecond, 128 << 20, "250ms", "256MiB", 2}, + {10 * time.Minute, 64 << 30, "1m0s", "16GiB", 2}, + {5 * time.Second, 4 << 30, "5s", "4GiB", 0}, + } + for _, tc := range cases { + h := newHarness(t, "script", func(c *SupervisorConfig) { + c.Timeout = tc.timeout + c.MemLimit = tc.mem + c.capOverride = fakeCap // the value is under test, not the cap + }) + if got := len(h.log.find("out of range; clamped")); got != tc.clampLogs { + t.Errorf("%v/%d: %d clamp warnings, want %d: %s", tc.timeout, tc.mem, got, tc.clampLogs, h.log.dump()) + } + if _, err := h.s.Materialize(context.Background(), script("echo:x")); err != nil { + t.Fatal(err) + } + got := h.log.find("materialize worker started", "timeout="+tc.wantT+" ", "mem_limit="+tc.wantM+" ", "mem_cap=fake(test)") + if len(got) != 1 { + t.Errorf("%v/%d: effective-values line missing: %s", tc.timeout, tc.mem, h.log.dump()) + } + } + // The production default names the platform mechanism. + h := newHarness(t, "script", func(c *SupervisorConfig) { c.capOverride = nil }) + if h.s.cap.mechanism != platformMemCap.mechanism { + t.Fatalf("mechanism %q", h.s.cap.mechanism) + } +} + +// --------------------------------------------------------------------------- +// the supervising process dies mid-job + +// TestSupervisorOrphanedWorkerExits: when the SUPERVISING process is killed +// while its worker is inside a job, the worker must not run on without its +// deadline kill and memory watchdog. Process tree: this test -> a helper +// parent running a Supervisor -> its worker, hung in a job. SIGKILL the helper +// parent; the worker must be gone within a second (Linux: Pdeathsig, and the +// ppid watch; macOS: the ppid watch; Windows: KILL_ON_JOB_CLOSE). +func TestSupervisorOrphanedWorkerExits(t *testing.T) { + parent := exec.Command(os.Args[0], "-test.run=^$") + parent.Env = append(os.Environ(), helperEnv+"=parent") + var stderr lockedBuffer + parent.Stderr = &stderr + out, err := parent.StdoutPipe() + if err != nil { + t.Fatal(err) + } + if err := parent.Start(); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = parent.Process.Kill(); _ = parent.Wait() }) + + lines := make(chan string, 4) + go func() { + sc := bufio.NewScanner(out) + for sc.Scan() { + lines <- sc.Text() + } + close(lines) + }() + next := func() string { + select { + case l, ok := <-lines: + if !ok { + t.Fatalf("helper parent ended early; stderr:\n%s", stderr.String()) + } + return l + case <-time.After(60 * time.Second): + t.Fatalf("helper parent silent; stderr:\n%s", stderr.String()) + } + return "" + } + var worker int + if _, err := fmt.Sscanf(next(), "worker %d", &worker); err != nil || worker <= 0 { + t.Fatalf("no worker pid: %v", err) + } + t.Cleanup(func() { + if processGone(worker) != nil { + if p, err := os.FindProcess(worker); err == nil { + _ = p.Kill() + } + } + }) + if l := next(); l != "hanging" { + t.Fatalf("got %q", l) + } + time.Sleep(300 * time.Millisecond) // the worker is now inside the hung job + if processGone(worker) == nil { + t.Fatal("premise: the worker must be alive before its parent dies") + } + + if err := parent.Process.Kill(); err != nil { + t.Fatal(err) + } + _ = parent.Wait() + killed := time.Now() + for processGone(worker) != nil { + if time.Since(killed) > 5*time.Second { + t.Fatalf("worker %d outlived its supervising process by 5s", worker) + } + time.Sleep(10 * time.Millisecond) + } + took := time.Since(killed) + t.Logf("worker %d gone %s after its parent was killed", worker, took.Round(time.Millisecond)) + if took > time.Second { + t.Fatalf("worker took %s to exit after its parent died; want under 1s", took) + } +} + +type lockedBuffer struct { + mu sync.Mutex + b strings.Builder +} + +func (l *lockedBuffer) Write(p []byte) (int, error) { + l.mu.Lock() + defer l.mu.Unlock() + return l.b.Write(p) +} + +func (l *lockedBuffer) String() string { + l.mu.Lock() + defer l.mu.Unlock() + return l.b.String() +} + +// --------------------------------------------------------------------------- +// 7. Close + +func TestSupervisorClose(t *testing.T) { + h := newHarness(t, "script", nil) + if _, err := h.s.Materialize(context.Background(), script("echo:x")); err != nil { + t.Fatal(err) + } + pid := h.lastPid(t) + if err := h.s.Close(); err != nil { + t.Fatal(err) + } + if err := processGone(pid); err != nil { + t.Fatalf("Close left the worker running: %v", err) + } + if err := h.s.Close(); err != nil { + t.Fatalf("second Close: %v", err) + } + if _, err := h.s.Materialize(context.Background(), script("echo:x")); !errors.Is(err, ErrClosed) { + t.Fatalf("after Close: %v", err) + } + if n := h.f.calls.Load(); n != 1 { + t.Fatalf("a job after Close spawned (%d spawns)", n) + } +} + +func TestSupervisorCloseDuringJob(t *testing.T) { + h := newHarness(t, "script", nil) + if _, err := h.s.Materialize(context.Background(), script("echo:x")); err != nil { + t.Fatal(err) + } + pid := h.lastPid(t) + done := make(chan error, 1) + go func() { + _, err := h.s.Materialize(context.Background(), script("hang")) + done <- err + }() + time.Sleep(200 * time.Millisecond) + _ = h.s.Close() + select { + case err := <-done: + if !errors.Is(err, ErrClosed) { + t.Fatalf("running job: %v, want ErrClosed", err) + } + case <-time.After(5 * time.Second): + t.Fatal("the running job did not end on Close") + } + if err := processGone(pid); err != nil { + t.Fatal(err) + } +} + +func TestSupervisorCloseBeforeFirstJob(t *testing.T) { + h := newHarness(t, "script", nil) + _ = h.s.Close() + if _, err := h.s.Materialize(context.Background(), script("echo:x")); !errors.Is(err, ErrClosed) { + t.Fatal(err) + } + if n := h.f.calls.Load(); n != 0 { + t.Fatalf("%d spawns", n) + } +} + +// --------------------------------------------------------------------------- +// idle exit (Dave's ruling on TASK-2198: a worker with no job for +// PAD_MATERIALIZE_IDLE_TIMEOUT stops, and the next job starts a new one) + +// waitIdleStopped polls until the supervisor has no worker. +func waitIdleStopped(t *testing.T, h *harness) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + h.s.mu.Lock() + c := h.s.child + h.s.mu.Unlock() + if c == nil && len(h.log.find("materialize worker stopped", "reason=idle")) > 0 { + return + } + time.Sleep(20 * time.Millisecond) + } + t.Fatalf("worker not stopped for idleness within 5s: %s", h.log.dump()) +} + +func TestSupervisorIdleExit(t *testing.T) { + h := newHarness(t, "script", func(c *SupervisorConfig) { c.idleExact = 200 * time.Millisecond }) + if md, err := h.s.Materialize(context.Background(), script("echo:a")); err != nil || md != "echo:a" { + t.Fatalf("first job: %q, %v", md, err) + } + pid := h.lastPid(t) + waitIdleStopped(t, h) + + // A clean stop: INFO, status 0, and no WARN stop line. + p := "pid=" + strconv.Itoa(pid) + if got := h.log.find("INFO materialize worker stopped", "reason=idle", "exited with status 0", p, "idle_for=200ms"); len(got) != 1 { + t.Fatalf("idle stop line: %s", h.log.dump()) + } + if w := h.log.find("WARN materialize worker stopped"); len(w) != 0 { + t.Fatalf("an idle stop logged as a kill: %q", w) + } + + // The next job starts a new worker, without any backoff wait. + if md, err := h.s.Materialize(context.Background(), script("echo:b")); err != nil || md != "echo:b" { + t.Fatalf("job after the idle stop: %q, %v", md, err) + } + if n := h.f.calls.Load(); n != 2 { + t.Fatalf("%d spawns; want 2 (lazy respawn after the idle stop)", n) + } + if got := h.clk.slept(); len(got) != 0 { + t.Fatalf("the respawn after an idle stop waited %v; want no backoff", got) + } + if h.lastPid(t) == pid { + t.Fatal("the job ran on the stopped worker") + } +} + +// An idle stop is not a death: the backoff streak neither grows nor resets. +// crash (streak 1) → idle stop → crash (streak 2) must wait 2s; had the idle +// stop counted it would be 4s. +func TestSupervisorIdleExitDoesNotTripBackoff(t *testing.T) { + h := newHarness(t, "script", func(c *SupervisorConfig) { + c.idleExact = 200 * time.Millisecond + c.BackoffBase = time.Second + c.BackoffMax = time.Minute + c.HealthyStretch = time.Hour + }) + if _, err := h.s.Materialize(context.Background(), script("exit")); !errors.Is(err, ErrChildDied) { + t.Fatal(err) + } + if _, err := h.s.Materialize(context.Background(), script("echo:x")); err != nil { + t.Fatal(err) + } + waitIdleStopped(t, h) + h.s.mu.Lock() + streak := h.s.streak + h.s.mu.Unlock() + if streak != 1 { + t.Fatalf("streak after an idle stop: %d, want 1 (only the crash)", streak) + } + if _, err := h.s.Materialize(context.Background(), script("exit")); !errors.Is(err, ErrChildDied) { + t.Fatal(err) + } + if _, err := h.s.Materialize(context.Background(), script("echo:y")); err != nil { + t.Fatal(err) + } + want := []time.Duration{time.Second, 2 * time.Second} + if got := h.clk.slept(); fmt.Sprint(got) != fmt.Sprint(want) { + t.Fatalf("backoff waits %v, want %v", got, want) + } +} + +// A job that arrives while an idle stop is in progress is not lost: it waits +// for the job slot, finds no worker, and runs on a new one. +func TestSupervisorJobDuringIdleStop(t *testing.T) { + result := make(chan error, 1) + var h *harness + h = newHarness(t, "script", func(c *SupervisorConfig) { + c.idleExact = 200 * time.Millisecond + c.onIdleStop = func() { + // The stop holds the slot and has forgotten the worker, which + // is still alive: exactly the window in question. + go func() { + md, err := h.s.Materialize(context.Background(), script("echo:during")) + if err == nil && md != "echo:during" { + err = fmt.Errorf("markdown %q", md) + } + result <- err + }() + time.Sleep(100 * time.Millisecond) // let it reach the slot + } + }) + if _, err := h.s.Materialize(context.Background(), script("echo:first")); err != nil { + t.Fatal(err) + } + select { + case err := <-result: + if err != nil { + t.Fatalf("the job that arrived during the idle stop: %v", err) + } + case <-time.After(10 * time.Second): + t.Fatalf("the job that arrived during the idle stop never finished: %s", h.log.dump()) + } + if n := h.f.calls.Load(); n != 2 { + t.Fatalf("%d spawns; want 2 (the job ran on a NEW worker)", n) + } + if got := h.clk.slept(); len(got) != 0 { + t.Fatalf("backoff waits %v; want none", got) + } +} + +// The idle timer can fire while a job is running, because it measures from +// the END of the previous job. The stop must not take that worker: it runs +// only holding the job slot, and a job still in flight holds it. (Closing +// stdin alone would let the job finish; the kill after the grace would not, +// so the grace is shorter than the job here.) +func TestSupervisorIdleTimerDuringJob(t *testing.T) { + h := newHarness(t, "script", func(c *SupervisorConfig) { + c.idleExact = 300 * time.Millisecond + c.idleGrace = 50 * time.Millisecond + }) + if _, err := h.s.Materialize(context.Background(), script("echo:x")); err != nil { + t.Fatal(err) + } + time.Sleep(200 * time.Millisecond) // the timer fires 100ms into the next job + if md, err := h.s.Materialize(context.Background(), script("sleep:400")); err != nil || md != "slept:400" { + t.Fatalf("a job running when the idle timer fired: %q, %v", md, err) + } + if n := h.f.calls.Load(); n != 1 { + t.Fatalf("%d spawns; the running job's worker was stopped under it", n) + } + // And the job's end re-armed the timer: the worker still goes idle later. + waitIdleStopped(t, h) +} + +// Close during an idle stop does not return while that worker still runs: +// the stop has already forgotten it, so Close must end and reap it itself. +func TestSupervisorCloseDuringIdleStop(t *testing.T) { + closed := make(chan struct{}) + var h *harness + h = newHarness(t, "script", func(c *SupervisorConfig) { + c.idleExact = 200 * time.Millisecond + c.idleGrace = time.Minute // the stop would otherwise wait on EOF + c.onIdleStop = func() { + go func() { + _ = h.s.Close() + close(closed) + }() + time.Sleep(100 * time.Millisecond) + } + }) + if _, err := h.s.Materialize(context.Background(), script("echo:x")); err != nil { + t.Fatal(err) + } + pid := h.lastPid(t) + select { + case <-closed: + case <-time.After(10 * time.Second): + t.Fatal("Close did not return") + } + if err := processGone(pid); err != nil { + t.Fatalf("Close returned while the idle-stopping worker still ran: %v", err) + } + // Its stop line names Close, not idleness (codex r2). + deadline := time.Now().Add(5 * time.Second) + for len(h.log.find("materialize worker stopped", "pid="+strconv.Itoa(pid))) == 0 && time.Now().Before(deadline) { + time.Sleep(20 * time.Millisecond) + } + if got := h.log.find("materialize worker stopped", "pid="+strconv.Itoa(pid), "reason=closed"); len(got) != 1 { + t.Fatalf("stop line for a worker Close ended mid idle stop: %s", h.log.dump()) + } +} + +func TestSupervisorNeverIdle(t *testing.T) { + h := newHarness(t, "script", func(c *SupervisorConfig) { c.IdleTimeout = NeverIdle }) + if _, err := h.s.Materialize(context.Background(), script("echo:x")); err != nil { + t.Fatal(err) + } + time.Sleep(300 * time.Millisecond) + h.s.mu.Lock() + c := h.s.child + h.s.mu.Unlock() + if c == nil || c.idleTimer != nil { + t.Fatalf("NeverIdle: child %v, idle timer armed %v", c != nil, c != nil && c.idleTimer != nil) + } +} + +func TestSupervisorIdleTimeoutConfig(t *testing.T) { + env := func(v string) func(string) string { + return func(k string) string { + if k == EnvIdleTimeout { + return v + } + return "" + } + } + for _, tc := range []struct { + in string + want time.Duration + warn bool + }{ + {"", 0, false}, {"0", NeverIdle, false}, {"10m", 10 * time.Minute, false}, + {"-5m", 0, true}, {"soon", 0, true}, + } { + sink := &logSink{} + if got := ConfigFromEnv(env(tc.in), slog.New(sink)).IdleTimeout; got != tc.want { + t.Errorf("%q: %v, want %v", tc.in, got, tc.want) + } + if warned := len(sink.warns()) > 0; warned != tc.warn { + t.Errorf("%q: warned %v, want %v: %s", tc.in, warned, tc.warn, sink.dump()) + } + } + for _, tc := range []struct { + in, want time.Duration + clamped bool + }{ + {0, DefaultIdleTimeout, false}, {NeverIdle, NeverIdle, false}, + {time.Second, MinIdleTimeout, true}, {48 * time.Hour, MaxIdleTimeout, true}, + {10 * time.Minute, 10 * time.Minute, false}, + } { + sink := &logSink{} + s := NewSupervisor(SupervisorConfig{IdleTimeout: tc.in, Logger: slog.New(sink), capOverride: fakeCap}) + if s.idle != tc.want { + t.Errorf("%v: effective %v, want %v", tc.in, s.idle, tc.want) + } + if got := len(sink.find(EnvIdleTimeout + " out of range; clamped")); (got == 1) != tc.clamped { + t.Errorf("%v: %d clamp warnings, clamped=%v", tc.in, got, tc.clamped) + } + _ = s.Close() + } +} diff --git a/internal/materialize/worker.go b/internal/materialize/worker.go index 0d9521142..f81fe8665 100644 --- a/internal/materialize/worker.go +++ b/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 { diff --git a/internal/materialize/worker_test.go b/internal/materialize/worker_test.go index 524c621f3..6bd383705 100644 --- a/internal/materialize/worker_test.go +++ b/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())