Rethinking cgroup Memory Enforcement

packages/engine-multi/README.md

@@ -81,6 +81,35 @@ engine.execute(plan)
 
 For a full list of events, see `src/events/ts` (the top-level API events are listed at the top)
 
+## Memory Limits
+
+The engine enforces two memory limits on each run:
+
+**Heap limit** (`memoryLimitMb`): sets V8's max old space size on the child process (and the worker thread's resource limits). If a run blows this, V8 aborts and the engine reports an OOMError. This only bounds the JavaScript heap - buffers and other native allocations don't count towards it.
+
+**cgroup limit** (`cgroupMemoryLimitMb`): a hard ceiling on the child process's total memory (including native allocations), enforced by the kernel through a cgroup v2 leaf. Each pooled child process is placed in its own cgroup under `cgroupParent` with `memory.max` set and swap disabled. If a run exceeds the ceiling, the kernel OOM-kills the child; the engine detects this from the cgroup's `memory.events` and reports an OOMError.
+
+The heap limit should sit below the cgroup limit, so that GC pressure kicks in first. The cgroup is a backstop for native (off-heap) memory, which the heap limit can't see.
+
+An OOMError carries a `source` property (`'heap'` or `'cgroup'`) saying which limit was breached.
+
+### cgroup setup
+
+The engine never provisions the cgroup hierarchy itself. The contract is that the worker process is **started inside** a writable, delegated cgroup v2 subtree; the engine then creates one leaf per child process within it. By default `cgroupParent` is the cgroup the worker was started in (read from `/proc/self/cgroup`), which makes the common environments work without configuration:
+
+| Environment | Setup required |
+| --- | --- |
+| Docker / K8s | Nothing to create - container processes are born in the container's cgroup. But the cgroup mount must be writable: unprivileged Docker mounts `/sys/fs/cgroup` read-only, so enforcement needs `--privileged` (or Podman, which mounts it read-write by default) |
+| systemd host | A unit with `User=openfn` and `Delegate=memory` - systemd creates the cgroup, chowns it to the user and starts the worker inside it |
+| Manual (no systemd) | As root: `mkdir` the cgroup and `chown` the dir, its `cgroup.procs` and `cgroup.subtree_control` to the worker user; then a root launcher writes its own pid into `cgroup.procs` before dropping privileges and `exec`ing node |
+| Local dev | Nothing - your cgroup isn't writable, so the engine warns once and falls back to heap-limit-only |
+
+If the contract isn't met (macOS, cgroup v1, no writable cgroup), the engine logs a warning once and falls back to heap-limit-only behaviour. All cgroup writes are confined to the delegated subtree: on first use the engine moves its own process into a `leader` leaf (cgroup v2 forbids delegating controllers from a populated cgroup) and enables the memory controller for its leaves.
+
+`cgroupParent` can be overridden, but pointing the worker at a subtree it wasn't started in generally requires root: the kernel only allows migrating a process if the writer has write access to the common ancestor's `cgroup.procs`.
+
+The ws-worker enables cgroup enforcement by default, with `run-memory + 128`mb of headroom for native allocations. Pass `--cgroup-memory 0` (or `WORKER_CGROUP_MEMORY_MB=0`) to disable it explicitly.
+
 ## Module Loader Whitelist
 
 A whitelist controls what modules a job is allowed to import. At the moment this is hardcoded in the Engine to modules starting with @openfn.

packages/engine-multi/docs/cgroup-spec.md

@@ -0,0 +1,134 @@
+# Spec: delegated cgroup memory enforcement
+
+Rework of the cgroup work on the `cgroup-oom-limit` branch (PR #1467).
+
+## Motivation
+
+The current implementation self-provisions the cgroup hierarchy: it walks down
+from `/sys/fs/cgroup`, enables the memory controller at each level, creates
+intermediate cgroups, and relocates any processes it finds in the root cgroup
+into a leader leaf. This:
+
+- requires root
+- mutates cgroup topology the engine does not own (on non-systemd hosts or
+  WSL2 it can relocate unrelated system processes)
+- makes the availability "probe" a side-effecting operation
+- conflicts with systemd's single-writer rule when creating cgroups directly
+  under the host root
+
+## Design
+
+Invert the responsibility. The engine never provisions the hierarchy; it
+consumes a **delegated** cgroup provided by the environment.
+
+**Contract**: the worker process must be *started inside* a writable cgroup v2
+subtree that has the memory controller available. Whoever starts the worker
+(systemd, Docker, a launcher script) is responsible for that. If the contract
+isn't met, the engine warns once and falls back to heap-limit-only behaviour.
+
+This works because processes are always born inside their supervisor's cgroup:
+
+| Environment | Setup required | How the worker ends up inside |
+| --- | --- | --- |
+| Docker / K8s | None to create; grant a writable cgroup mount (`--privileged`, or Podman/crun which mounts rw by default). Unprivileged Docker mounts `/sys/fs/cgroup` read-only → clean fallback | Container processes are born in the container's cgroup (= namespace root) |
+| systemd host | Unit with `User=openfn` and `Delegate=memory` (systemd creates the cgroup, chowns it to the user) | Service is born in its delegated cgroup |
+| Manual (no systemd) | Root: `mkdir` the cgroup; `chown` the dir, `cgroup.procs` and `cgroup.subtree_control` to the worker user | Root launcher writes `$$` into `cgroup.procs`, drops privileges, `exec`s node |
+| Local dev | None | Own cgroup isn't writable → warn + fallback |
+
+**Default parent = the worker's own cgroup**, resolved from
+`/proc/self/cgroup` (the `0::<path>` line, joined onto `/sys/fs/cgroup`).
+This makes all rows above work with identical logic and no configuration.
+`cgroupParent` / `--cgroup-parent` remains as an override, but note the kernel
+migration rule: moving a pid between cgroups requires write access to the
+*common ancestor's* `cgroup.procs`, so pointing the worker at a subtree it
+doesn't live inside only works when running as root.
+
+### Startup sequence (all writes confined to the parent)
+
+1. **Detect**: Linux; parent dir exists; `<parent>/cgroup.controllers` lists
+   `memory`; parent is writable. No mutation. Any failure → warn once,
+   fall back.
+2. **Prepare** (first use): move the processes currently in the parent (within
+   a delegated subtree these can only be the worker's own) into a
+   `<parent>/leader` leaf — required by the no-internal-processes rule — then
+   write `+memory` to `<parent>/cgroup.subtree_control`. Failure → warn,
+   fall back.
+3. **Per child fork**: `mkdir <parent>/run-<pid>`; write `memory.max`
+   (`cgroupMemoryLimitMb`); write `memory.swap.max = 0` (best-effort); write
+   the pid to `cgroup.procs` last. Failure → warn, run continues heap-only.
+4. **On unexpected child exit**: read the leaf's `memory.events`; if
+   `oom_kill > 0`, reject with `OOMError('cgroup')` before falling back to the
+   stderr heap-message scrape. (Unchanged from the current branch.)
+5. **Cleanup**: remove the leaf on kill/timeout/destroy, retrying briefly on
+   `EBUSY`/`ENOTEMPTY`. (Unchanged.)
+
+## Code changes
+
+### `packages/engine-multi/src/worker/cgroup.ts`
+
+- Delete `ensureParent` and the root-walking/controller-enabling logic;
+  `CGROUP_ROOT` is no longer written to, only used to resolve paths.
+- Rescope `moveProcsToLeader` to operate only on the configured parent (step 2
+  above); it must never be called on a directory outside the parent.
+- Add `resolveSelfCgroup(): string | null` — parse `/proc/self/cgroup`,
+  return the absolute path of the v2 cgroup, null if unavailable. This
+  replaces `DEFAULT_CGROUP_PARENT` as the default.
+- `detect` becomes read-only (step 1); the prepare step (2) runs lazily on
+  first successful detection. Cache the result per parent as now.
+- `createChildCgroup`, `hasOomKill`, `removeChildCgroup`: unchanged.
+
+### `packages/engine-multi/src/worker/pool.ts`
+
+- No behavioural change; the default parent now comes from
+  `resolveSelfCgroup()` rather than a hardcoded path.
+
+### `packages/ws-worker`
+
+- `--cgroup-memory` unchanged: defaults to `run-memory + 128`, `0` disables
+  (documented). Default-on is now safe because the failure mode is a warning,
+  not host mutation.
+- `--cgroup-parent` help text: default is the worker's own cgroup; overriding
+  it generally requires root (common-ancestor rule).
+
+### Error propagation (decide: this PR or follow-up)
+
+`OOMError.source` (`'heap' | 'cgroup'`) currently dies at
+`lifecycle.error()`, which only forwards `type`/`message`/`severity` — so
+Lightning cannot distinguish the two cases. If the PR's "identifier" claim is
+meant to reach Lightning, forward `source` through the `WORKFLOW_ERROR` event
+and into the exit reason; otherwise soften the claim to "identifiable in
+worker logs".
+
+## Tests
+
+- **Gate the enforcement tests on `isCgroupV2Available(...)`, not
+  `os.platform() === 'linux'`.** As written they run on any Linux host —
+  including CI runners and dev machines where cgroups are unavailable — and
+  `blowNativeMemory` then allocates unbounded native memory with no limit of
+  any kind.
+- Unit tests (any platform): `resolveSelfCgroup` parsing (fixture strings);
+  detection returns false without side effects when the parent is missing,
+  read-only, or lacks the memory controller; existing null-handle tolerance
+  tests stand.
+- Enforcement tests (writable-cgroup hosts only): kernel OOM-kill surfaces
+  `OOMError` with `source: 'cgroup'`; pool restores the slot and keeps
+  working. Runnable locally via `docker run --privileged` (recipe stays in
+  the test header); consider a privileged CI job later.
+
+## Docs
+
+- Rewrite the README "Memory Limits" section around the delegation contract
+  and the setup table above.
+- Deployment note: unprivileged Docker/K8s mounts `/sys/fs/cgroup` read-only,
+  so enforcement needs `--privileged` (or Podman) — verify against the actual
+  deployment infra before publishing ops guidance.
+
+## Out of scope / open questions
+
+- `memory.oom.group=1` on leaves — irrelevant while each leaf holds exactly
+  one process; revisit if jobs ever spawn subprocesses.
+- Tuning the `+128`mb headroom default: children are reused across runs, so
+  native memory accumulates toward the ceiling; legitimate runs near the heap
+  limit could be kernel-killed. May want a larger default or per-deployment
+  guidance.
+- Changeset still needed on the PR.

packages/engine-multi/src/api.ts

@@ -58,6 +58,8 @@ const createAPI = async function (
 
     maxWorkers: options.maxWorkers,
     memoryLimitMb: options.memoryLimitMb || DEFAULT_MEMORY_LIMIT,
+    cgroupMemoryLimitMb: options.cgroupMemoryLimitMb,
+    cgroupParent: options.cgroupParent,
     runTimeoutMs: options.runTimeoutMs,
 
     statePropsToRemove: options.statePropsToRemove ?? [

packages/engine-multi/src/api/call-worker.ts

@@ -13,6 +13,8 @@ type WorkerOptions = {
   env?: any;
   timeout?: number; // ms
   memoryLimitMb?: number;
+  cgroupMemoryLimitMb?: number; // hard cgroup v2 memory.max ceiling
+  cgroupParent?: string; // parent cgroup for per-child leaves
   proxyStdout?: boolean; // print internal stdout to console
 };
 
@@ -27,6 +29,8 @@ export default function initWorkers(
     env = {},
     maxWorkers = 5,
     memoryLimitMb,
+    cgroupMemoryLimitMb,
+    cgroupParent,
     proxyStdout = false,
   } = options;
 
@@ -36,6 +40,8 @@ export default function initWorkers(
       maxWorkers,
       env,
       memoryLimitMb,
+      cgroupMemoryLimitMb,
+      cgroupParent,
       proxyStdout,
     },
     logger

packages/engine-multi/src/api/lifecycle.ts

@@ -155,5 +155,7 @@ export const error = (
     message: error.message || error.toString(),
     // default to exception because if we don't know, it's our fault
     severity: error.severity || 'exception',
+    // @ts-ignore for OOM errors, say which limit was breached (heap/cgroup)
+    source: error.source,
   });
 };

packages/engine-multi/src/engine.ts

@@ -75,6 +75,8 @@ export type EngineOptions = {
   maxWorkers?: number;
   memoryLimitMb?: number;
   stateLimitMb?: number;
+  cgroupMemoryLimitMb?: number;
+  cgroupParent?: string;
   payloadLimitMb?: number;
   logPayloadLimitMb?: number;
   repoDir: string;
@@ -142,6 +144,8 @@ const createEngine = async (
     {
       maxWorkers: options.maxWorkers,
       memoryLimitMb: defaultMemoryLimit,
+      cgroupMemoryLimitMb: options.cgroupMemoryLimitMb,
+      cgroupParent: options.cgroupParent,
       proxyStdout: options.proxyStdout,
     },
     options.logger

packages/engine-multi/src/errors.ts

@@ -71,14 +71,21 @@ export class AutoinstallError extends EngineError {
   }
 }
 
+// Where an OOM kill came from:
+// - 'heap'   the V8 heap limit (thread resourceLimits or --max-old-space-size)
+// - 'cgroup' the OS cgroup memory.max ceiling (kernel SIGKILL)
+export type OOMSource = 'heap' | 'cgroup';
+
 export class OOMError extends EngineError {
   severity = 'kill';
   name = 'OOMError';
   message;
+  source: OOMSource;
 
-  constructor() {
+  constructor(source: OOMSource = 'heap') {
     super();
 
+    this.source = source;
     this.message = `Run exceeded maximum memory usage`;
   }
 }

packages/engine-multi/src/events.ts

@@ -71,6 +71,9 @@ export interface WorkflowErrorPayload extends ExternalEvent {
   type: string;
   message: string;
   severity: string;
+  // where the error originated; 'engine' generally, but OOM errors narrow
+  // this to the limit that was breached ('heap' or 'cgroup')
+  source?: string;
 }
 
 export interface JobStartPayload extends ExternalEvent {

packages/engine-multi/src/test/worker-functions.ts

@@ -119,6 +119,17 @@ const tasks = {
     // Array(1e9).fill('mario')
   },
 
+  // Allocate memory OUTSIDE the V8 heap. Buffer.alloc lives in native memory,
+  // so --max-old-space-size can't catch this — only an OS-level limit (a
+  // cgroup memory.max ceiling) will. Zero-filling forces the pages resident so
+  // they count against the cgroup's accounting.
+  blowNativeMemory: async () => {
+    const chunks = [];
+    while (true) {
+      chunks.push(Buffer.alloc(10 * 1024 * 1024, 1)); // 10mb, resident
+    }
+  },
+
   // Some useful code
   // const stats = v8.getHeapStatistics();
   // console.log(

packages/engine-multi/src/util/serialize-error.ts

@@ -4,6 +4,7 @@ export default (error: any) => {
     name: error.name,
     type: error.type,
     subtype: error.subtype,
+    source: error.source,
     severity: error.severity || 'crash',
   };
 };