Rethinking cgroup Memory Enforcement

packages/engine-multi/README.md

@@ -81,6 +81,26 @@ 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 a memory limits on each run via `memoryLimitMb`. There are two modes of enforcement:
+
+**Heap limit**: 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. This is enabled by default.
+
+**cgroup limit**: 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.
+
+An OOMError carries a `source` property (`'heap'` or `'cgroup'`) saying which limit was breached.
+
+### cgroup setup
+
+The engine does not provision the cgroup hierarchy itself. Rather, the host runtime must be started within a writable, cgroup v2 subtree.
+
+The engine creates one leaf per child process within it, applying the memory limit to each.
+
+cgroups must be enabled explicitly: the are off by default because unless properly configured, an OOM exception can kill the parenting process group hierarchy.
+
+See Worker documentation for notes about how to start the engine with cgroups enabled.
+
 ## 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/ava.config.cjs

@@ -2,5 +2,11 @@ const baseConfig = require('../../ava.config');
 
 module.exports = {
   ...baseConfig,
-  files: ['test/**/*.test.ts'],
+  files: [
+    'test/**/*.test.ts',
+
+    // Enforcement tests must be rub in an isolated cgroup
+    // via pnpm test:cgroup
+    '!test/worker/cgroup-enforcement.test.ts',
+  ],
 };

packages/engine-multi/package.json

@@ -6,6 +6,7 @@
   "type": "module",
   "scripts": {
     "test": "pnpm ava --serial",
+    "test:cgroup": "systemd-run --user --scope --unit=cgroup-test -p Delegate=yes -p MemoryMax=2G -- pnpm exec ava test/worker/cgroup-enforcement.test.ts",
     "test:types": "pnpm tsc --noEmit --project tsconfig.json",
     "test:mem": "NODE_OPTIONS=\"--max-old-space-size=90 --experimental-vm-modules\" pnpm exec tsx test/memtest.ts",
     "build": "tsup --config ./tsup.config.js",

packages/engine-multi/src/api.ts

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

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

@@ -1,5 +1,5 @@
 import { Logger } from '@openfn/logger';
-import createPool from '../worker/pool';
+import createPool, { MemoryEnforcement } from '../worker/pool';
 import { CallWorker } from '../types';
 
 // All events coming out of the worker need to include a type key
@@ -13,6 +13,7 @@ type WorkerOptions = {
   env?: any;
   timeout?: number; // ms
   memoryLimitMb?: number;
+  memoryEnforcement?: MemoryEnforcement;
   proxyStdout?: boolean; // print internal stdout to console
 };
 
@@ -27,6 +28,7 @@ export default function initWorkers(
     env = {},
     maxWorkers = 5,
     memoryLimitMb,
+    memoryEnforcement,
     proxyStdout = false,
   } = options;
 
@@ -36,6 +38,7 @@ export default function initWorkers(
       maxWorkers,
       env,
       memoryLimitMb,
+      memoryEnforcement,
       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

@@ -13,6 +13,7 @@ import {
   WORKFLOW_START,
 } from './events';
 import initWorkers from './api/call-worker';
+import type { MemoryEnforcement } from './worker/pool';
 import createState from './util/create-state';
 import execute from './api/execute';
 import validateWorker from './api/validate-worker';
@@ -74,6 +75,7 @@ export type EngineOptions = {
   logger: Logger;
   maxWorkers?: number;
   memoryLimitMb?: number;
+  memoryEnforcement?: MemoryEnforcement;
   stateLimitMb?: number;
   payloadLimitMb?: number;
   logPayloadLimitMb?: number;
@@ -142,6 +144,7 @@ const createEngine = async (
     {
       maxWorkers: options.maxWorkers,
       memoryLimitMb: defaultMemoryLimit,
+      memoryEnforcement: options.memoryEnforcement,
       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',
   };
 };