From e1cab41415ff4c267b6bf0ed7a3538918fde5f83 Mon Sep 17 00:00:00 2001 From: Farhan Yahaya Date: Mon, 29 Jun 2026 07:39:30 +0000 Subject: [PATCH 01/15] feat: init cgroups constraints --- packages/engine-multi/src/api.ts | 2 + packages/engine-multi/src/api/call-worker.ts | 6 + packages/engine-multi/src/engine.ts | 4 + .../engine-multi/src/test/worker-functions.ts | 11 + packages/engine-multi/src/worker/cgroup.ts | 262 ++++++++++++++++++ packages/engine-multi/src/worker/pool.ts | 54 ++++ .../test/worker/cgroup-enforcement.test.ts | 51 ++++ .../engine-multi/test/worker/cgroup.test.ts | 73 +++++ packages/ws-worker/src/start.ts | 4 + packages/ws-worker/src/util/cli.ts | 20 ++ 10 files changed, 487 insertions(+) create mode 100644 packages/engine-multi/src/worker/cgroup.ts create mode 100644 packages/engine-multi/test/worker/cgroup-enforcement.test.ts create mode 100644 packages/engine-multi/test/worker/cgroup.test.ts diff --git a/packages/engine-multi/src/api.ts b/packages/engine-multi/src/api.ts index 29450b16a..783ac47d0 100644 --- a/packages/engine-multi/src/api.ts +++ b/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 ?? [ diff --git a/packages/engine-multi/src/api/call-worker.ts b/packages/engine-multi/src/api/call-worker.ts index c3a062407..8a8cbdea5 100644 --- a/packages/engine-multi/src/api/call-worker.ts +++ b/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 diff --git a/packages/engine-multi/src/engine.ts b/packages/engine-multi/src/engine.ts index 38467b97c..2dc5ce545 100644 --- a/packages/engine-multi/src/engine.ts +++ b/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 diff --git a/packages/engine-multi/src/test/worker-functions.ts b/packages/engine-multi/src/test/worker-functions.ts index 8a0068827..a56f9d5be 100644 --- a/packages/engine-multi/src/test/worker-functions.ts +++ b/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( diff --git a/packages/engine-multi/src/worker/cgroup.ts b/packages/engine-multi/src/worker/cgroup.ts new file mode 100644 index 000000000..1aec9e33c --- /dev/null +++ b/packages/engine-multi/src/worker/cgroup.ts @@ -0,0 +1,262 @@ +// cgroup v2 memory enforcement for pooled child processes. +// +// The pool already limits the V8 heap via --max-old-space-size, but that does +// not bound native allocations, buffers or overall RSS. On Linux hosts that +// expose a writable cgroup v2 hierarchy we additionally place each child in its +// own leaf cgroup with a hard `memory.max` ceiling, so the kernel OOM-kills a +// runaway run before it can starve its neighbours. +// +// This is strictly best-effort: on non-Linux hosts, cgroup v1, or when we lack +// the permissions/delegation to create cgroups, every function degrades to a +// no-op and the caller falls back to heap-limit-only behaviour. + +import fs from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; +import type { Logger } from '@openfn/logger'; + +const CGROUP_ROOT = '/sys/fs/cgroup'; + +// Leaf cgroup into which we relocate any processes that sit in a cgroup we need +// to delegate the memory controller from (see enableMemoryController). +const LEADER_NAME = 'openfn-leader'; + +// Default parent cgroup under which per-child leaf cgroups are created. +// Must be a path the worker process can write to (typically requires cgroup +// delegation in a container, or running as root). +export const DEFAULT_CGROUP_PARENT = path.join(CGROUP_ROOT, 'openfn'); + +export type CgroupHandle = { + path: string; // absolute path to the leaf cgroup directory + eventsPath: string; // absolute path to the leaf's memory.events file +}; + +// Cache availability per parent so we only probe the filesystem (and warn) once. +const availabilityCache: Record = {}; + +// True if a cgroup interface file (e.g. cgroup.controllers, subtree_control) +// lists the given whitespace-separated token. +const fileLists = (file: string, token: string) => + fs.readFileSync(file, 'utf8').split(/\s+/).includes(token); + +// --------------------------------------------------------------------------- +// Availability & controller delegation (private) +// --------------------------------------------------------------------------- + +// Relocate every process in `dir` into a dedicated leader leaf cgroup, so `dir` +// itself becomes empty and its controllers can be delegated. This is the same +// move systemd/runc make; it includes the worker's own process when `dir` is +// the container's cgroup namespace root. +const moveProcsToLeader = (dir: string, logger: Logger) => { + const leader = path.join(dir, LEADER_NAME); + if (!fs.existsSync(leader)) { + fs.mkdirSync(leader); + } + const procs = fs + .readFileSync(path.join(dir, 'cgroup.procs'), 'utf8') + .trim() + .split('\n') + .filter(Boolean); + for (const pid of procs) { + try { + fs.writeFileSync(path.join(leader, 'cgroup.procs'), pid); + } catch (e) { + // Some processes (e.g. kernel threads) can't be moved; skip them. + logger.debug(`cgroup: could not move pid ${pid} into leader: ${ + (e as Error).message + }`); + } + } +}; + +// Enable the memory controller for the children of `dir` (idempotent). +const enableMemoryController = (dir: string, logger: Logger) => { + const subtree = path.join(dir, 'cgroup.subtree_control'); + if (fileLists(subtree, 'memory')) { + return; + } + try { + fs.writeFileSync(subtree, '+memory'); + } catch (e) { + // EBUSY means `dir` still holds member processes (cgroup v2's "no internal + // processes" rule forbids delegating from a populated cgroup). This is the + // common containerised case where the worker lives in the namespace root: + // move those processes into a leader leaf, then retry the delegation. + if ((e as NodeJS.ErrnoException).code === 'EBUSY') { + logger.debug( + `cgroup: ${dir} is populated; relocating processes to delegate memory` + ); + moveProcsToLeader(dir, logger); + fs.writeFileSync(subtree, '+memory'); + } else { + throw e; + } + } +}; + +// Ensure the parent cgroup exists and delegates the memory controller all the +// way down from the cgroup root, so leaf cgroups created under it can set +// memory.max. Throws on any failure (caught by the caller). +const ensureParent = (parent: string, logger: Logger) => { + const rel = path.relative(CGROUP_ROOT, parent); + if (!rel || rel.startsWith('..') || path.isAbsolute(rel)) { + throw new Error(`cgroup parent ${parent} is not under ${CGROUP_ROOT}`); + } + + // Delegate the memory controller from the root down to (and including) the + // parent, creating each intermediate cgroup as needed. Any process sitting in + // a cgroup we delegate from (notably the worker itself, in the namespace + // root) is relocated to a leader leaf first; processes otherwise only live in + // the leaves below `parent`. + enableMemoryController(CGROUP_ROOT, logger); + let dir = CGROUP_ROOT; + for (const segment of rel.split(path.sep).filter(Boolean)) { + dir = path.join(dir, segment); + if (!fs.existsSync(dir)) { + fs.mkdirSync(dir); + } + enableMemoryController(dir, logger); + } +}; + +const detect = (parent: string, logger: Logger): boolean => { + if (os.platform() !== 'linux') { + return false; + } + + // cgroup v2 (the unified hierarchy) exposes cgroup.controllers at the root; + // cgroup v1 does not. + const rootControllers = path.join(CGROUP_ROOT, 'cgroup.controllers'); + if (!fs.existsSync(rootControllers)) { + return false; + } + + try { + if (!fileLists(rootControllers, 'memory')) { + return false; + } + ensureParent(parent, logger); + return true; + } catch (e) { + logger.debug( + `cgroup: setup of parent ${parent} failed: ${(e as Error).message}` + ); + return false; + } +}; + +// --------------------------------------------------------------------------- +// Public API +// --------------------------------------------------------------------------- + +// Returns true when cgroup v2 memory enforcement can be used under `parent`. +// Caches the result and warns (once per parent) when unavailable. +export const isCgroupV2Available = ( + parent: string, + logger: Logger +): boolean => { + if (parent in availabilityCache) { + return availabilityCache[parent]; + } + + const available = detect(parent, logger); + availabilityCache[parent] = available; + if (!available) { + logger.warn( + 'cgroup: v2 memory enforcement unavailable on this host; ' + + 'falling back to heap-limit only' + ); + } + return available; +}; + +// Create a leaf cgroup for `pid`, set its hard memory ceiling and move the +// process into it. Returns a handle, or null on any failure (best-effort). +export const createChildCgroup = ( + parent: string, + pid: number, + limitBytes: number, + logger: Logger +): CgroupHandle | null => { + const dir = path.join(parent, `run-${pid}`); + try { + if (!fs.existsSync(dir)) { + fs.mkdirSync(dir); + } + fs.writeFileSync(path.join(dir, 'memory.max'), String(limitBytes)); + // Stop the process escaping the limit via swap. The swap controller may be + // absent, so this is best-effort. + try { + fs.writeFileSync(path.join(dir, 'memory.swap.max'), '0'); + } catch (e) { + // no swap controller; ignore + } + // Moving the pid in must come last: a cgroup with member processes cannot + // have its controllers reconfigured. + fs.writeFileSync(path.join(dir, 'cgroup.procs'), String(pid)); + logger.debug( + `cgroup: constrained process ${pid} to ${limitBytes} bytes at ${dir}` + ); + return { path: dir, eventsPath: path.join(dir, 'memory.events') }; + } catch (e) { + logger.warn( + `cgroup: failed to constrain process ${pid}: ${(e as Error).message}` + ); + return null; + } +}; + +// Returns true if the kernel OOM-killed anything in this cgroup. Used to tell a +// cgroup memory kill (a bare SIGKILL with no V8 message) apart from other crashes. +export const hasOomKill = (handle: CgroupHandle): boolean => { + try { + const content = fs.readFileSync(handle.eventsPath, 'utf8'); + for (const line of content.split('\n')) { + const [key, value] = line.trim().split(/\s+/); + if ( + (key === 'oom_kill' || key === 'oom_group_kill') && + parseInt(value, 10) > 0 + ) { + return true; + } + } + } catch (e) { + // events file gone (cgroup already removed) or unreadable + } + return false; +}; + +// Remove a leaf cgroup. The cgroup can only be removed once empty, so we retry +// a handful of times while the killed process is reaped. Best-effort and +// non-blocking. +export const removeChildCgroup = ( + handle: CgroupHandle | null, + logger: Logger, + attempts = 10 +) => { + if (!handle) { + return; + } + try { + fs.rmdirSync(handle.path); + } catch (e) { + const err = e as NodeJS.ErrnoException; + if ( + attempts > 0 && + (err.code === 'EBUSY' || err.code === 'ENOTEMPTY') + ) { + setTimeout(() => removeChildCgroup(handle, logger, attempts - 1), 50); + return; + } + if (err.code !== 'ENOENT') { + logger.debug(`cgroup: could not remove ${handle.path}: ${err.message}`); + } + } +}; + +// For tests: clear the cached availability probes. +export const _resetAvailabilityCache = () => { + for (const key of Object.keys(availabilityCache)) { + delete availabilityCache[key]; + } +}; diff --git a/packages/engine-multi/src/worker/pool.ts b/packages/engine-multi/src/worker/pool.ts index 48a268803..c3f3c0e0d 100644 --- a/packages/engine-multi/src/worker/pool.ts +++ b/packages/engine-multi/src/worker/pool.ts @@ -13,6 +13,14 @@ import { import { HANDLED_EXIT_CODE } from '../events'; import { Logger } from '@openfn/logger'; import type { PayloadLimits } from './thread/runtime'; +import { + CgroupHandle, + DEFAULT_CGROUP_PARENT, + createChildCgroup, + hasOomKill, + isCgroupV2Available, + removeChildCgroup, +} from './cgroup'; export type PoolOptions = { capacity?: number; // defaults to 5 @@ -20,6 +28,12 @@ export type PoolOptions = { env?: Record; // default environment for workers memoryLimitMb?: number; // --max-old-space-size for child processes + // Hard memory.max ceiling (mb) applied to each child via a cgroup v2 leaf. + // Best-effort: ignored on hosts without a writable cgroup v2 hierarchy. + cgroupMemoryLimitMb?: number; + // Parent cgroup under which per-child leaf cgroups are created. + cgroupParent?: string; + proxyStdout?: boolean; // print internal stdout to console }; @@ -80,6 +94,21 @@ function createPool(script: string, options: PoolOptions = {}, logger: Logger) { // Keep track of all the workers we created const allWorkers: Record = {}; + // cgroup v2 leaf per child (keyed by pid), when memory enforcement is enabled + const cgroupParent = options.cgroupParent || DEFAULT_CGROUP_PARENT; + const cgroupEnabled = + !!options.cgroupMemoryLimitMb && + isCgroupV2Available(cgroupParent, logger); + const cgroups: Record = {}; + + // Tear down a child's leaf cgroup once it has been killed (best-effort). + const cleanupCgroup = (worker: ChildProcess | false) => { + if (worker && worker.pid && cgroups[worker.pid]) { + removeChildCgroup(cgroups[worker.pid], logger); + delete cgroups[worker.pid]; + } + }; + const init = (maybeChild: ChildProcess | false) => { let child: ChildProcess; if (!maybeChild) { @@ -107,6 +136,17 @@ function createPool(script: string, options: PoolOptions = {}, logger: Logger) { logger.debug('pool: Created new child process', child.pid); allWorkers[child.pid!] = child; + + // Place the child in its own cgroup with a hard memory ceiling. The leaf + // lives for the lifetime of the (reused) child and is removed when it dies. + if (cgroupEnabled && child.pid) { + cgroups[child.pid] = createChildCgroup( + cgroupParent, + child.pid, + options.cgroupMemoryLimitMb! * 1024 * 1024, + logger + ); + } } else { child = maybeChild as ChildProcess; logger.debug('pool: Using existing child process', child.pid); @@ -153,6 +193,17 @@ function createPool(script: string, options: PoolOptions = {}, logger: Logger) { logger.debug(`pool: Worker exited unexpectedly with code ${code}`); clearTimeout(timeout); + // A cgroup memory kill is a bare SIGKILL with no V8 message, so check + // the cgroup's OOM counter before falling back to scraping stderr. + const handle = worker.pid ? cgroups[worker.pid] : null; + if (handle && hasOomKill(handle)) { + killWorker(worker); + // restore a placeholder to the queue + finish(false); + reject(new OOMError()); + return; + } + // Read the stderr stream from the worked to see if this looks like an OOM error const rl = readline.createInterface({ input: worker.stderr!, @@ -265,6 +316,7 @@ function createPool(script: string, options: PoolOptions = {}, logger: Logger) { if (worker) { logger.debug('pool: destroying worker ', worker.pid); worker.kill(); + cleanupCgroup(worker); delete allWorkers[worker.pid!]; } }; @@ -305,6 +357,7 @@ function createPool(script: string, options: PoolOptions = {}, logger: Logger) { const worker = pool.pop(); if (worker) { killPromises.push(waitForWorkerExit(worker)); + cleanupCgroup(worker); delete allWorkers[worker.pid!]; } } @@ -312,6 +365,7 @@ function createPool(script: string, options: PoolOptions = {}, logger: Logger) { if (immediate) { Object.values(allWorkers).forEach((worker) => { killPromises.push(waitForWorkerExit(worker, 1)); + cleanupCgroup(worker); delete allWorkers[worker.pid!]; }); } diff --git a/packages/engine-multi/test/worker/cgroup-enforcement.test.ts b/packages/engine-multi/test/worker/cgroup-enforcement.test.ts new file mode 100644 index 000000000..aace0eca5 --- /dev/null +++ b/packages/engine-multi/test/worker/cgroup-enforcement.test.ts @@ -0,0 +1,51 @@ +import test from 'ava'; +import os from 'node:os'; +import path from 'node:path'; +import { createMockLogger } from '@openfn/logger'; + +import createPool from '../../src/worker/pool'; + +// These tests prove the kernel actually OOM-kills a runaway run via the +// cgroup's memory.max ceiling. That only works on Linux with a writable +// cgroup v2 hierarchy (e.g. a privileged container), so they are skipped +// everywhere else — including CI on macOS. +// +// docker run --rm -it --privileged -v "$PWD":/kit -w /kit node:24 bash +// corepack enable && pnpm install && pnpm --filter @openfn/engine-multi build +// cd packages/engine-multi && pnpm ava test/worker/cgroup-enforcement.test.ts +const linuxOnly = os.platform() === 'linux' ? test.serial : test.serial.skip; + +const workerPath = path.resolve('dist/test/worker-functions.js'); +const logger = createMockLogger(); + +// Ceiling above Node's baseline RSS but low enough that blowNativeMemory +// crosses it almost immediately. memoryLimitMb is deliberately left unset so +// the V8 heap limit can't be what kills the run — only the cgroup can. +const cgroupMemoryLimitMb = 200; + +linuxOnly( + 'cgroup OOM-kills a run that exceeds memory.max and surfaces OOMError', + async (t) => { + const pool = createPool(workerPath, { cgroupMemoryLimitMb }, logger); + + await t.throwsAsync(() => pool.exec('blowNativeMemory', []), { + name: 'OOMError', + }); + + await pool.destroy(); + } +); + +linuxOnly('pool recovers after a cgroup OOM kill', async (t) => { + const pool = createPool(workerPath, { cgroupMemoryLimitMb }, logger); + + await t.throwsAsync(() => pool.exec('blowNativeMemory', []), { + name: 'OOMError', + }); + + // The pool should restore the dead worker's slot and keep working. + const result = await pool.exec('test', []); + t.is(result, 42); + + await pool.destroy(); +}); diff --git a/packages/engine-multi/test/worker/cgroup.test.ts b/packages/engine-multi/test/worker/cgroup.test.ts new file mode 100644 index 000000000..28a779338 --- /dev/null +++ b/packages/engine-multi/test/worker/cgroup.test.ts @@ -0,0 +1,73 @@ +import test from 'ava'; +import os from 'node:os'; +import { createMockLogger } from '@openfn/logger'; + +import { + DEFAULT_CGROUP_PARENT, + createChildCgroup, + hasOomKill, + isCgroupV2Available, + removeChildCgroup, + _resetAvailabilityCache, +} from '../../src/worker/cgroup'; + +const logger = createMockLogger(); + +const isLinux = os.platform() === 'linux'; + +test.beforeEach(() => { + _resetAvailabilityCache(); +}); + +test('DEFAULT_CGROUP_PARENT lives under the cgroup root', (t) => { + t.is(DEFAULT_CGROUP_PARENT, '/sys/fs/cgroup/openfn'); +}); + +// cgroups don't exist on macOS/Windows, so the whole module must no-op there. +if (!isLinux) { + test('isCgroupV2Available returns false on non-linux hosts', (t) => { + t.false(isCgroupV2Available(DEFAULT_CGROUP_PARENT, logger)); + }); + + test('availability probe is cached and only warns once', (t) => { + const l = createMockLogger('test', { level: 'debug' }); + isCgroupV2Available('/sys/fs/cgroup/test-cache', l); + isCgroupV2Available('/sys/fs/cgroup/test-cache', l); + isCgroupV2Available('/sys/fs/cgroup/test-cache', l); + + const warnings = l._history.filter((h: any) => h[0] === 'warn'); + t.is(warnings.length, 1); + }); +} + +test('createChildCgroup returns null when the parent is not writable', (t) => { + const handle = createChildCgroup( + '/this/path/does/not/exist', + 12345, + 256 * 1024 * 1024, + logger + ); + t.is(handle, null); +}); + +test('hasOomKill returns false when the events file is missing', (t) => { + t.false( + hasOomKill({ + path: '/no/such/cgroup', + eventsPath: '/no/such/cgroup/memory.events', + }) + ); +}); + +test('removeChildCgroup tolerates a null handle', (t) => { + t.notThrows(() => removeChildCgroup(null, logger)); +}); + +test('removeChildCgroup tolerates a non-existent cgroup', (t) => { + t.notThrows(() => + removeChildCgroup( + { path: '/no/such/cgroup', eventsPath: '/no/such/cgroup/memory.events' }, + logger + ) + ); +}); diff --git a/packages/ws-worker/src/start.ts b/packages/ws-worker/src/start.ts index 4e08e2601..f37f8b2f1 100644 --- a/packages/ws-worker/src/start.ts +++ b/packages/ws-worker/src/start.ts @@ -112,6 +112,10 @@ if (args.mock) { repoDir: args.repoDir, memoryLimitMb: args.runMemory, stateLimitMb: args.stateMemory, + // Hard cgroup ceiling sits above the V8 heap limit so GC fires first; + // default to run-memory + 128mb of headroom for native/buffer allocations. + cgroupMemoryLimitMb: args.cgroupMemory ?? (args.runMemory ?? 500) + 128, + cgroupParent: args.cgroupParent, maxWorkers: effectiveCapacity, statePropsToRemove: args.statePropsToRemove, runTimeoutMs: args.maxRunDurationSeconds * 1000, diff --git a/packages/ws-worker/src/util/cli.ts b/packages/ws-worker/src/util/cli.ts index f2850a383..e54e86f6a 100644 --- a/packages/ws-worker/src/util/cli.ts +++ b/packages/ws-worker/src/util/cli.ts @@ -16,6 +16,8 @@ type Args = { batchLimit?: number; batchLogs: boolean; capacity: number; + cgroupMemory?: number; + cgroupParent?: string; workloops?: string; claimTimeoutSeconds?: number; collectionsUrl?: string; @@ -82,6 +84,8 @@ export default function parseArgs(argv: string[]): Args { WORKER_BATCH_LIMIT, WORKER_BATCH_LOGS, WORKER_CAPACITY, + WORKER_CGROUP_MEMORY_MB, + WORKER_CGROUP_PARENT, WORKER_CLAIM_TIMEOUT_SECONDS, WORKER_COLLECTIONS_URL, WORKER_COLLECTIONS_VERSION, @@ -231,6 +235,16 @@ export default function parseArgs(argv: string[]): Args { 'Maximum memory allocated to a single run, in mb. Env: WORKER_MAX_PAYLOAD_MB', type: 'number', }) + .option('cgroup-memory', { + description: + 'Hard memory ceiling (mb) enforced per run via a cgroup v2 leaf. Linux only; falls back to heap-limit only when unavailable. Defaults to run-memory + 128. Env: WORKER_CGROUP_MEMORY_MB', + type: 'number', + }) + .option('cgroup-parent', { + description: + 'Parent cgroup path under which per-run leaf cgroups are created. Must be writable (cgroup delegation/root). Default /sys/fs/cgroup/openfn. Env: WORKER_CGROUP_PARENT', + type: 'string', + }) .option('max-run-duration-seconds', { alias: 't', @@ -339,6 +353,12 @@ export default function parseArgs(argv: string[]): Args { (WORKER_MAX_STATE_MEMORY_MB ? parseInt(WORKER_MAX_STATE_MEMORY_MB, 10) : undefined), + cgroupMemory: + args.cgroupMemory ?? + (WORKER_CGROUP_MEMORY_MB ? parseInt(WORKER_CGROUP_MEMORY_MB) : undefined), + cgroupParent: setArg(args.cgroupParent, WORKER_CGROUP_PARENT) as + | string + | undefined, payloadMemory: setArg(args.payloadMemory, WORKER_MAX_PAYLOAD_MB, 10), logPayloadMemory: setArg( args.logPayloadMemory, From 9af5b7e138e5ab100cd7bd58bfde536f351ce0de Mon Sep 17 00:00:00 2001 From: Farhan Yahaya Date: Mon, 29 Jun 2026 08:30:44 +0000 Subject: [PATCH 02/15] feat: determine where OOM came from --- packages/engine-multi/src/errors.ts | 9 ++++++++- packages/engine-multi/src/util/serialize-error.ts | 1 + packages/engine-multi/src/worker/pool.ts | 11 ++++++++++- .../test/worker/cgroup-enforcement.test.ts | 3 ++- packages/engine-multi/test/worker/pool.test.ts | 12 +++++++++--- 5 files changed, 30 insertions(+), 6 deletions(-) diff --git a/packages/engine-multi/src/errors.ts b/packages/engine-multi/src/errors.ts index ca88cce83..9ac2e4eef 100644 --- a/packages/engine-multi/src/errors.ts +++ b/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`; } } diff --git a/packages/engine-multi/src/util/serialize-error.ts b/packages/engine-multi/src/util/serialize-error.ts index 73b8a3e53..48718be73 100644 --- a/packages/engine-multi/src/util/serialize-error.ts +++ b/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', }; }; diff --git a/packages/engine-multi/src/worker/pool.ts b/packages/engine-multi/src/worker/pool.ts index c3f3c0e0d..dcda57d08 100644 --- a/packages/engine-multi/src/worker/pool.ts +++ b/packages/engine-multi/src/worker/pool.ts @@ -197,10 +197,14 @@ function createPool(script: string, options: PoolOptions = {}, logger: Logger) { // the cgroup's OOM counter before falling back to scraping stderr. const handle = worker.pid ? cgroups[worker.pid] : null; if (handle && hasOomKill(handle)) { + logger.error( + `pool: worker ${worker.pid} was killed by the OS for exceeding ` + + `its cgroup memory limit (${options.cgroupMemoryLimitMb}mb)` + ); killWorker(worker); // restore a placeholder to the queue finish(false); - reject(new OOMError()); + reject(new OOMError('cgroup')); return; } @@ -214,6 +218,9 @@ function createPool(script: string, options: PoolOptions = {}, logger: Logger) { if (worker.stderr && worker.stderr?.readableLength > 0) { for await (const line of rl) { if (line.match(/JavaScript heap out of memory/)) { + logger.error( + `pool: worker ${worker.pid} exceeded the V8 heap limit` + ); killWorker(worker); // restore a placeholder to the queue finish(false); @@ -301,6 +308,8 @@ function createPool(script: string, options: PoolOptions = {}, logger: Logger) { // @ts-ignore e.severity = evt.error.severity; e.name = evt.error.name; + // @ts-ignore preserve the OOM source ('heap'/'cgroup') across IPC + e.source = evt.error.source; reject(e); finish(worker); diff --git a/packages/engine-multi/test/worker/cgroup-enforcement.test.ts b/packages/engine-multi/test/worker/cgroup-enforcement.test.ts index aace0eca5..7054097c2 100644 --- a/packages/engine-multi/test/worker/cgroup-enforcement.test.ts +++ b/packages/engine-multi/test/worker/cgroup-enforcement.test.ts @@ -28,9 +28,10 @@ linuxOnly( async (t) => { const pool = createPool(workerPath, { cgroupMemoryLimitMb }, logger); - await t.throwsAsync(() => pool.exec('blowNativeMemory', []), { + const err = await t.throwsAsync(() => pool.exec('blowNativeMemory', []), { name: 'OOMError', }); + t.is((err as any).source, 'cgroup'); await pool.destroy(); } diff --git a/packages/engine-multi/test/worker/pool.test.ts b/packages/engine-multi/test/worker/pool.test.ts index 5a3b8d9dc..6c02ccc0d 100644 --- a/packages/engine-multi/test/worker/pool.test.ts +++ b/packages/engine-multi/test/worker/pool.test.ts @@ -179,9 +179,11 @@ test('throw if memory limit is exceeded', async (t) => { try { await pool.exec('blowMemory', [], { memoryLimitMb: 100 }); + t.fail('expected the run to OOM'); } catch (e: any) { t.is(e.message, 'Run exceeded maximum memory usage'); t.is(e.name, 'OOMError'); + t.is(e.source, 'heap'); } }); @@ -200,9 +202,13 @@ test('child process should not have --max-old-space-size when memoryLimitMb is n test('pool recovers after process-level OOM', async (t) => { const pool = createPool(workerPath, { memoryLimitMb: 50 }, logger); - await t.throwsAsync(() => pool.exec('blowMemory', [], { memoryLimitMb: 20 }), { - name: 'OOMError', - }); + const err = await t.throwsAsync( + () => pool.exec('blowMemory', [], { memoryLimitMb: 20 }), + { + name: 'OOMError', + } + ); + t.is((err as any).source, 'heap'); // Pool should still be functional after the OOM const result = await pool.exec('test', [42]); From 71c4967884dbf7c355e1429437b0d653bf1c89e7 Mon Sep 17 00:00:00 2001 From: Joe Clark Date: Thu, 16 Jul 2026 16:49:28 +0100 Subject: [PATCH 03/15] light refactor --- packages/engine-multi/src/worker/cgroup.ts | 260 ++++++++++----------- 1 file changed, 122 insertions(+), 138 deletions(-) diff --git a/packages/engine-multi/src/worker/cgroup.ts b/packages/engine-multi/src/worker/cgroup.ts index 1aec9e33c..dec682ec4 100644 --- a/packages/engine-multi/src/worker/cgroup.ts +++ b/packages/engine-multi/src/worker/cgroup.ts @@ -1,15 +1,9 @@ -// cgroup v2 memory enforcement for pooled child processes. -// -// The pool already limits the V8 heap via --max-old-space-size, but that does -// not bound native allocations, buffers or overall RSS. On Linux hosts that -// expose a writable cgroup v2 hierarchy we additionally place each child in its -// own leaf cgroup with a hard `memory.max` ceiling, so the kernel OOM-kills a -// runaway run before it can starve its neighbours. -// -// This is strictly best-effort: on non-Linux hosts, cgroup v1, or when we lack -// the permissions/delegation to create cgroups, every function degrades to a -// no-op and the caller falls back to heap-limit-only behaviour. - +/** + * cgroup v2 memory enforcement for pooled child processes. + * + * cgroup level memory enforcement allows us to enforce memory limits at the kernel level, + * improving our ability to OOMKill runs. + */ import fs from 'node:fs'; import os from 'node:os'; import path from 'node:path'; @@ -17,10 +11,6 @@ import type { Logger } from '@openfn/logger'; const CGROUP_ROOT = '/sys/fs/cgroup'; -// Leaf cgroup into which we relocate any processes that sit in a cgroup we need -// to delegate the memory controller from (see enableMemoryController). -const LEADER_NAME = 'openfn-leader'; - // Default parent cgroup under which per-child leaf cgroups are created. // Must be a path the worker process can write to (typically requires cgroup // delegation in a container, or running as root). @@ -31,124 +21,6 @@ export type CgroupHandle = { eventsPath: string; // absolute path to the leaf's memory.events file }; -// Cache availability per parent so we only probe the filesystem (and warn) once. -const availabilityCache: Record = {}; - -// True if a cgroup interface file (e.g. cgroup.controllers, subtree_control) -// lists the given whitespace-separated token. -const fileLists = (file: string, token: string) => - fs.readFileSync(file, 'utf8').split(/\s+/).includes(token); - -// --------------------------------------------------------------------------- -// Availability & controller delegation (private) -// --------------------------------------------------------------------------- - -// Relocate every process in `dir` into a dedicated leader leaf cgroup, so `dir` -// itself becomes empty and its controllers can be delegated. This is the same -// move systemd/runc make; it includes the worker's own process when `dir` is -// the container's cgroup namespace root. -const moveProcsToLeader = (dir: string, logger: Logger) => { - const leader = path.join(dir, LEADER_NAME); - if (!fs.existsSync(leader)) { - fs.mkdirSync(leader); - } - const procs = fs - .readFileSync(path.join(dir, 'cgroup.procs'), 'utf8') - .trim() - .split('\n') - .filter(Boolean); - for (const pid of procs) { - try { - fs.writeFileSync(path.join(leader, 'cgroup.procs'), pid); - } catch (e) { - // Some processes (e.g. kernel threads) can't be moved; skip them. - logger.debug(`cgroup: could not move pid ${pid} into leader: ${ - (e as Error).message - }`); - } - } -}; - -// Enable the memory controller for the children of `dir` (idempotent). -const enableMemoryController = (dir: string, logger: Logger) => { - const subtree = path.join(dir, 'cgroup.subtree_control'); - if (fileLists(subtree, 'memory')) { - return; - } - try { - fs.writeFileSync(subtree, '+memory'); - } catch (e) { - // EBUSY means `dir` still holds member processes (cgroup v2's "no internal - // processes" rule forbids delegating from a populated cgroup). This is the - // common containerised case where the worker lives in the namespace root: - // move those processes into a leader leaf, then retry the delegation. - if ((e as NodeJS.ErrnoException).code === 'EBUSY') { - logger.debug( - `cgroup: ${dir} is populated; relocating processes to delegate memory` - ); - moveProcsToLeader(dir, logger); - fs.writeFileSync(subtree, '+memory'); - } else { - throw e; - } - } -}; - -// Ensure the parent cgroup exists and delegates the memory controller all the -// way down from the cgroup root, so leaf cgroups created under it can set -// memory.max. Throws on any failure (caught by the caller). -const ensureParent = (parent: string, logger: Logger) => { - const rel = path.relative(CGROUP_ROOT, parent); - if (!rel || rel.startsWith('..') || path.isAbsolute(rel)) { - throw new Error(`cgroup parent ${parent} is not under ${CGROUP_ROOT}`); - } - - // Delegate the memory controller from the root down to (and including) the - // parent, creating each intermediate cgroup as needed. Any process sitting in - // a cgroup we delegate from (notably the worker itself, in the namespace - // root) is relocated to a leader leaf first; processes otherwise only live in - // the leaves below `parent`. - enableMemoryController(CGROUP_ROOT, logger); - let dir = CGROUP_ROOT; - for (const segment of rel.split(path.sep).filter(Boolean)) { - dir = path.join(dir, segment); - if (!fs.existsSync(dir)) { - fs.mkdirSync(dir); - } - enableMemoryController(dir, logger); - } -}; - -const detect = (parent: string, logger: Logger): boolean => { - if (os.platform() !== 'linux') { - return false; - } - - // cgroup v2 (the unified hierarchy) exposes cgroup.controllers at the root; - // cgroup v1 does not. - const rootControllers = path.join(CGROUP_ROOT, 'cgroup.controllers'); - if (!fs.existsSync(rootControllers)) { - return false; - } - - try { - if (!fileLists(rootControllers, 'memory')) { - return false; - } - ensureParent(parent, logger); - return true; - } catch (e) { - logger.debug( - `cgroup: setup of parent ${parent} failed: ${(e as Error).message}` - ); - return false; - } -}; - -// --------------------------------------------------------------------------- -// Public API -// --------------------------------------------------------------------------- - // Returns true when cgroup v2 memory enforcement can be used under `parent`. // Caches the result and warns (once per parent) when unavailable. export const isCgroupV2Available = ( @@ -241,10 +113,7 @@ export const removeChildCgroup = ( fs.rmdirSync(handle.path); } catch (e) { const err = e as NodeJS.ErrnoException; - if ( - attempts > 0 && - (err.code === 'EBUSY' || err.code === 'ENOTEMPTY') - ) { + if (attempts > 0 && (err.code === 'EBUSY' || err.code === 'ENOTEMPTY')) { setTimeout(() => removeChildCgroup(handle, logger, attempts - 1), 50); return; } @@ -260,3 +129,118 @@ export const _resetAvailabilityCache = () => { delete availabilityCache[key]; } }; + +// Leaf cgroup into which we relocate any processes that sit in a cgroup we need +// to delegate the memory controller from (see enableMemoryController). +const LEADER_NAME = 'openfn-leader'; + +// Cache availability per parent so we only probe the filesystem (and warn) once. +const availabilityCache: Record = {}; + +// True if a cgroup interface file (e.g. cgroup.controllers, subtree_control) +// lists the given whitespace-separated token. +function fileLists(file: string, token: string) { + return fs.readFileSync(file, 'utf8').split(/\s+/).includes(token); +} + +// Relocate every process in `dir` into a dedicated leader leaf cgroup, so `dir` +// itself becomes empty and its controllers can be delegated. This is the same +// move systemd/runc make; it includes the worker's own process when `dir` is +// the container's cgroup namespace root. +function moveProcsToLeader(dir: string, logger: Logger) { + const leader = path.join(dir, LEADER_NAME); + if (!fs.existsSync(leader)) { + fs.mkdirSync(leader); + } + const procs = fs + .readFileSync(path.join(dir, 'cgroup.procs'), 'utf8') + .trim() + .split('\n') + .filter(Boolean); + for (const pid of procs) { + try { + fs.writeFileSync(path.join(leader, 'cgroup.procs'), pid); + } catch (e) { + // Some processes (e.g. kernel threads) can't be moved; skip them. + logger.debug( + `cgroup: could not move pid ${pid} into leader: ${(e as Error).message}` + ); + } + } +} + +// Enable the memory controller for the children of `dir` (idempotent). +function enableMemoryController(dir: string, logger: Logger) { + const subtree = path.join(dir, 'cgroup.subtree_control'); + if (fileLists(subtree, 'memory')) { + return; + } + try { + fs.writeFileSync(subtree, '+memory'); + } catch (e) { + // EBUSY means `dir` still holds member processes (cgroup v2's "no internal + // processes" rule forbids delegating from a populated cgroup). This is the + // common containerised case where the worker lives in the namespace root: + // move those processes into a leader leaf, then retry the delegation. + if ((e as NodeJS.ErrnoException).code === 'EBUSY') { + logger.debug( + `cgroup: ${dir} is populated; relocating processes to delegate memory` + ); + moveProcsToLeader(dir, logger); + fs.writeFileSync(subtree, '+memory'); + } else { + throw e; + } + } +} + +// Ensure the parent cgroup exists and delegates the memory controller all the +// way down from the cgroup root, so leaf cgroups created under it can set +// memory.max. Throws on any failure (caught by the caller). +function ensureParent(parent: string, logger: Logger) { + const rel = path.relative(CGROUP_ROOT, parent); + if (!rel || rel.startsWith('..') || path.isAbsolute(rel)) { + throw new Error(`cgroup parent ${parent} is not under ${CGROUP_ROOT}`); + } + + // Delegate the memory controller from the root down to (and including) the + // parent, creating each intermediate cgroup as needed. Any process sitting in + // a cgroup we delegate from (notably the worker itself, in the namespace + // root) is relocated to a leader leaf first; processes otherwise only live in + // the leaves below `parent`. + enableMemoryController(CGROUP_ROOT, logger); + let dir = CGROUP_ROOT; + for (const segment of rel.split(path.sep).filter(Boolean)) { + dir = path.join(dir, segment); + if (!fs.existsSync(dir)) { + fs.mkdirSync(dir); + } + enableMemoryController(dir, logger); + } +} + +function detect(parent: string, logger: Logger): boolean { + if (os.platform() !== 'linux') { + return false; + } + + // cgroup v2 (the unified hierarchy) exposes cgroup.controllers at the root; + // cgroup v1 does not. + const rootControllers = path.join(CGROUP_ROOT, 'cgroup.controllers'); + if (!fs.existsSync(rootControllers)) { + return false; + } + + try { + if (!fileLists(rootControllers, 'memory')) { + return false; + } + ensureParent(parent, logger); + return true; + } catch (e) { + logger.debug( + `cgroup: setup of parent ${parent} failed: ${(e as Error).message}` + ); + return false; + } +} From a1211e3f6e75beb8e3945eec32db1f98df8fa11e Mon Sep 17 00:00:00 2001 From: Joe Clark Date: Thu, 16 Jul 2026 17:15:13 +0100 Subject: [PATCH 04/15] docs --- packages/engine-multi/README.md | 16 ++++++++++++++++ packages/ws-worker/src/util/cli.ts | 2 +- 2 files changed, 17 insertions(+), 1 deletion(-) diff --git a/packages/engine-multi/README.md b/packages/engine-multi/README.md index bc7b1b9a9..532b2fa57 100644 --- a/packages/engine-multi/README.md +++ b/packages/engine-multi/README.md @@ -81,6 +81,22 @@ 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` (default `/sys/fs/cgroup/openfn`) 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 enforcement is best-effort. It needs Linux with a writable cgroup v2 hierarchy - typically root inside a container with cgroup delegation. Anywhere else (macOS, cgroup v1, unprivileged processes - which in practice includes most local dev machines) the engine logs a warning once and falls back to heap-limit-only behaviour. As part of setup, the engine may relocate processes at the cgroup root into a leader leaf so the memory controller can be delegated (the same move systemd and runc make). + +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. diff --git a/packages/ws-worker/src/util/cli.ts b/packages/ws-worker/src/util/cli.ts index e54e86f6a..ca3bc9e14 100644 --- a/packages/ws-worker/src/util/cli.ts +++ b/packages/ws-worker/src/util/cli.ts @@ -237,7 +237,7 @@ export default function parseArgs(argv: string[]): Args { }) .option('cgroup-memory', { description: - 'Hard memory ceiling (mb) enforced per run via a cgroup v2 leaf. Linux only; falls back to heap-limit only when unavailable. Defaults to run-memory + 128. Env: WORKER_CGROUP_MEMORY_MB', + 'Hard memory ceiling (mb) enforced per run via a cgroup v2 leaf. Linux only; falls back to heap-limit only when unavailable. Defaults to run-memory + 128. Set to 0 to disable. Env: WORKER_CGROUP_MEMORY_MB', type: 'number', }) .option('cgroup-parent', { From dac182d47a03eb4e99b810b9fe492fb4e6ee0fa2 Mon Sep 17 00:00:00 2001 From: Joe Clark Date: Thu, 16 Jul 2026 17:52:58 +0100 Subject: [PATCH 05/15] re-implement cgroup stuff --- packages/engine-multi/README.md | 17 +- packages/engine-multi/docs/cgroup-spec.md | 134 ++++++++++++++ packages/engine-multi/src/api/lifecycle.ts | 2 + packages/engine-multi/src/events.ts | 3 + packages/engine-multi/src/worker/cgroup.ts | 173 +++++++++++------- packages/engine-multi/src/worker/pool.ts | 13 +- .../test/worker/cgroup-enforcement.test.ts | 25 ++- .../engine-multi/test/worker/cgroup.test.ts | 71 +++++-- packages/ws-worker/src/util/cli.ts | 2 +- 9 files changed, 344 insertions(+), 96 deletions(-) create mode 100644 packages/engine-multi/docs/cgroup-spec.md diff --git a/packages/engine-multi/README.md b/packages/engine-multi/README.md index 532b2fa57..8ecd33bdd 100644 --- a/packages/engine-multi/README.md +++ b/packages/engine-multi/README.md @@ -87,13 +87,26 @@ 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` (default `/sys/fs/cgroup/openfn`) 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. +**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 enforcement is best-effort. It needs Linux with a writable cgroup v2 hierarchy - typically root inside a container with cgroup delegation. Anywhere else (macOS, cgroup v1, unprivileged processes - which in practice includes most local dev machines) the engine logs a warning once and falls back to heap-limit-only behaviour. As part of setup, the engine may relocate processes at the cgroup root into a leader leaf so the memory controller can be delegated (the same move systemd and runc make). +### 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. diff --git a/packages/engine-multi/docs/cgroup-spec.md b/packages/engine-multi/docs/cgroup-spec.md new file mode 100644 index 000000000..2bbd3bc0c --- /dev/null +++ b/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::` 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; `/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 + `/leader` leaf — required by the no-internal-processes rule — then + write `+memory` to `/cgroup.subtree_control`. Failure → warn, + fall back. +3. **Per child fork**: `mkdir /run-`; 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. diff --git a/packages/engine-multi/src/api/lifecycle.ts b/packages/engine-multi/src/api/lifecycle.ts index 817140899..bd1a6254f 100644 --- a/packages/engine-multi/src/api/lifecycle.ts +++ b/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, }); }; diff --git a/packages/engine-multi/src/events.ts b/packages/engine-multi/src/events.ts index f9a81c32d..7d62230b2 100644 --- a/packages/engine-multi/src/events.ts +++ b/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 { diff --git a/packages/engine-multi/src/worker/cgroup.ts b/packages/engine-multi/src/worker/cgroup.ts index dec682ec4..4540e7160 100644 --- a/packages/engine-multi/src/worker/cgroup.ts +++ b/packages/engine-multi/src/worker/cgroup.ts @@ -3,6 +3,12 @@ * * cgroup level memory enforcement allows us to enforce memory limits at the kernel level, * improving our ability to OOMKill runs. + * + * The engine never provisions the cgroup hierarchy itself. The contract is that + * the worker process is *started inside* a delegated, writable cgroup - by a + * systemd unit with Delegate=, by a container runtime, or by a launcher script + * (see docs/cgroup-spec.md). When the contract isn't met, every function here + * degrades to a no-op and the caller falls back to heap-limit-only behaviour. */ import fs from 'node:fs'; import os from 'node:os'; @@ -11,32 +17,80 @@ import type { Logger } from '@openfn/logger'; const CGROUP_ROOT = '/sys/fs/cgroup'; -// Default parent cgroup under which per-child leaf cgroups are created. -// Must be a path the worker process can write to (typically requires cgroup -// delegation in a container, or running as root). -export const DEFAULT_CGROUP_PARENT = path.join(CGROUP_ROOT, 'openfn'); - export type CgroupHandle = { path: string; // absolute path to the leaf cgroup directory eventsPath: string; // absolute path to the leaf's memory.events file }; +// Extract the cgroup v2 path (the "0::" entry) from a /proc/self/cgroup +// listing and resolve it against the cgroup mount. Returns null if the process +// has no v2 cgroup (ie, a pure cgroup v1 host) +export const parseProcSelfCgroup = (content: string): string | null => { + for (const line of content.split('\n')) { + const match = line.match(/^0::(.+)$/); + if (match) { + // strip the trailing slash left by the namespace root case ("0::/") + return path.join(CGROUP_ROOT, match[1]).replace(/(.)\/$/, '$1'); + } + } + return null; +}; + +// The cgroup this process was started in. This is the default parent for +// per-child leaves: under systemd Delegate= or in a container it is exactly +// the subtree that was handed to us. Returns null when it can't be determined +export const resolveSelfCgroup = (): string | null => { + if (os.platform() !== 'linux') { + return null; + } + try { + const self = parseProcSelfCgroup( + fs.readFileSync('/proc/self/cgroup', 'utf8') + ); + // If we were started inside a leader leaf (eg, restarted from within an + // already-prepared cgroup), the delegated parent is the enclosing cgroup + if (self && path.basename(self) === LEADER_NAME) { + return path.dirname(self); + } + return self; + } catch (e) { + return null; + } +}; + // Returns true when cgroup v2 memory enforcement can be used under `parent`. -// Caches the result and warns (once per parent) when unavailable. +// On first success this also prepares the parent (delegates the memory +// controller to its leaves). Caches the result and warns (once per parent) +// when unavailable export const isCgroupV2Available = ( - parent: string, + parent: string | null, logger: Logger ): boolean => { - if (parent in availabilityCache) { - return availabilityCache[parent]; + const key = parent ?? ''; + if (key in availabilityCache) { + return availabilityCache[key]; + } + + let available = false; + if (parent && detect(parent, logger)) { + try { + prepare(parent, logger); + available = true; + } catch (e) { + logger.debug( + `cgroup: could not delegate memory controller in ${parent}: ${ + (e as Error).message + }` + ); + } } - const available = detect(parent, logger); - availabilityCache[parent] = available; + availabilityCache[key] = available; if (!available) { logger.warn( - 'cgroup: v2 memory enforcement unavailable on this host; ' + - 'falling back to heap-limit only' + 'cgroup: v2 memory enforcement unavailable; falling back to heap-limit only. ' + + 'To enable it, start the worker inside a writable, delegated cgroup ' + + '(see the engine-multi README)' ); } return available; @@ -114,7 +168,10 @@ export const removeChildCgroup = ( } catch (e) { const err = e as NodeJS.ErrnoException; if (attempts > 0 && (err.code === 'EBUSY' || err.code === 'ENOTEMPTY')) { - setTimeout(() => removeChildCgroup(handle, logger, attempts - 1), 50); + setTimeout( + () => removeChildCgroup(handle, logger, attempts - 1), + 50 + ).unref(); return; } if (err.code !== 'ENOENT') { @@ -130,9 +187,10 @@ export const _resetAvailabilityCache = () => { } }; -// Leaf cgroup into which we relocate any processes that sit in a cgroup we need -// to delegate the memory controller from (see enableMemoryController). -const LEADER_NAME = 'openfn-leader'; +// Leaf cgroup into which the worker relocates the processes in its own cgroup +// (typically just itself), so the memory controller can be delegated to the +// run leaves (see prepare) +const LEADER_NAME = 'leader'; // Cache availability per parent so we only probe the filesystem (and warn) once. const availabilityCache: Record = {}; @@ -143,17 +201,17 @@ function fileLists(file: string, token: string) { return fs.readFileSync(file, 'utf8').split(/\s+/).includes(token); } -// Relocate every process in `dir` into a dedicated leader leaf cgroup, so `dir` -// itself becomes empty and its controllers can be delegated. This is the same -// move systemd/runc make; it includes the worker's own process when `dir` is -// the container's cgroup namespace root. -function moveProcsToLeader(dir: string, logger: Logger) { - const leader = path.join(dir, LEADER_NAME); +// Relocate every process in `parent` into a leader leaf, so `parent` itself +// becomes empty and its controllers can be delegated. Within a delegated +// subtree the only member processes are our own (typically just the worker, +// when it was started inside `parent`) +function moveProcsToLeader(parent: string, logger: Logger) { + const leader = path.join(parent, LEADER_NAME); if (!fs.existsSync(leader)) { fs.mkdirSync(leader); } const procs = fs - .readFileSync(path.join(dir, 'cgroup.procs'), 'utf8') + .readFileSync(path.join(parent, 'cgroup.procs'), 'utf8') .trim() .split('\n') .filter(Boolean); @@ -161,7 +219,8 @@ function moveProcsToLeader(dir: string, logger: Logger) { try { fs.writeFileSync(path.join(leader, 'cgroup.procs'), pid); } catch (e) { - // Some processes (e.g. kernel threads) can't be moved; skip them. + // A process may exit between the read and the move; skip it and let the + // subsequent subtree_control write decide whether the parent is empty logger.debug( `cgroup: could not move pid ${pid} into leader: ${(e as Error).message}` ); @@ -169,24 +228,24 @@ function moveProcsToLeader(dir: string, logger: Logger) { } } -// Enable the memory controller for the children of `dir` (idempotent). -function enableMemoryController(dir: string, logger: Logger) { - const subtree = path.join(dir, 'cgroup.subtree_control'); +// Enable the memory controller for the leaves of `parent` (idempotent). +// cgroup v2's "no internal processes" rule forbids delegating controllers +// from a populated cgroup, and in the recommended setup the worker itself +// lives in `parent` - so on EBUSY, move its processes into a leader leaf +// and retry +function prepare(parent: string, logger: Logger) { + const subtree = path.join(parent, 'cgroup.subtree_control'); if (fileLists(subtree, 'memory')) { return; } try { fs.writeFileSync(subtree, '+memory'); } catch (e) { - // EBUSY means `dir` still holds member processes (cgroup v2's "no internal - // processes" rule forbids delegating from a populated cgroup). This is the - // common containerised case where the worker lives in the namespace root: - // move those processes into a leader leaf, then retry the delegation. if ((e as NodeJS.ErrnoException).code === 'EBUSY') { logger.debug( - `cgroup: ${dir} is populated; relocating processes to delegate memory` + `cgroup: ${parent} is populated; relocating processes to a leader leaf` ); - moveProcsToLeader(dir, logger); + moveProcsToLeader(parent, logger); fs.writeFileSync(subtree, '+memory'); } else { throw e; @@ -194,52 +253,34 @@ function enableMemoryController(dir: string, logger: Logger) { } } -// Ensure the parent cgroup exists and delegates the memory controller all the -// way down from the cgroup root, so leaf cgroups created under it can set -// memory.max. Throws on any failure (caught by the caller). -function ensureParent(parent: string, logger: Logger) { - const rel = path.relative(CGROUP_ROOT, parent); - if (!rel || rel.startsWith('..') || path.isAbsolute(rel)) { - throw new Error(`cgroup parent ${parent} is not under ${CGROUP_ROOT}`); - } - - // Delegate the memory controller from the root down to (and including) the - // parent, creating each intermediate cgroup as needed. Any process sitting in - // a cgroup we delegate from (notably the worker itself, in the namespace - // root) is relocated to a leader leaf first; processes otherwise only live in - // the leaves below `parent`. - enableMemoryController(CGROUP_ROOT, logger); - let dir = CGROUP_ROOT; - for (const segment of rel.split(path.sep).filter(Boolean)) { - dir = path.join(dir, segment); - if (!fs.existsSync(dir)) { - fs.mkdirSync(dir); - } - enableMemoryController(dir, logger); - } -} - +// Read-only check that `parent` is a usable delegated cgroup: a cgroup v2 +// directory under the cgroup mount, writable by this process, with the memory +// controller available function detect(parent: string, logger: Logger): boolean { if (os.platform() !== 'linux') { return false; } - // cgroup v2 (the unified hierarchy) exposes cgroup.controllers at the root; - // cgroup v1 does not. - const rootControllers = path.join(CGROUP_ROOT, 'cgroup.controllers'); - if (!fs.existsSync(rootControllers)) { + const rel = path.relative(CGROUP_ROOT, parent); + if (rel.startsWith('..') || path.isAbsolute(rel)) { + logger.debug(`cgroup: parent ${parent} is not under ${CGROUP_ROOT}`); return false; } try { - if (!fileLists(rootControllers, 'memory')) { + // cgroup v2 exposes cgroup.controllers in every cgroup; v1 does not + const controllers = path.join(parent, 'cgroup.controllers'); + if (!fileLists(controllers, 'memory')) { + logger.debug(`cgroup: memory controller not available in ${parent}`); return false; } - ensureParent(parent, logger); + // We need to create leaf dirs and move pids around within the parent + fs.accessSync(parent, fs.constants.W_OK); + fs.accessSync(path.join(parent, 'cgroup.procs'), fs.constants.W_OK); return true; } catch (e) { logger.debug( - `cgroup: setup of parent ${parent} failed: ${(e as Error).message}` + `cgroup: parent ${parent} is not usable: ${(e as Error).message}` ); return false; } diff --git a/packages/engine-multi/src/worker/pool.ts b/packages/engine-multi/src/worker/pool.ts index dcda57d08..199d2c920 100644 --- a/packages/engine-multi/src/worker/pool.ts +++ b/packages/engine-multi/src/worker/pool.ts @@ -15,11 +15,11 @@ import { Logger } from '@openfn/logger'; import type { PayloadLimits } from './thread/runtime'; import { CgroupHandle, - DEFAULT_CGROUP_PARENT, createChildCgroup, hasOomKill, isCgroupV2Available, removeChildCgroup, + resolveSelfCgroup, } from './cgroup'; export type PoolOptions = { @@ -31,7 +31,9 @@ export type PoolOptions = { // Hard memory.max ceiling (mb) applied to each child via a cgroup v2 leaf. // Best-effort: ignored on hosts without a writable cgroup v2 hierarchy. cgroupMemoryLimitMb?: number; - // Parent cgroup under which per-child leaf cgroups are created. + // Parent cgroup under which per-child leaf cgroups are created. Defaults to + // the cgroup this process was started in; overriding it generally requires + // root (the kernel checks the common ancestor's cgroup.procs on migration) cgroupParent?: string; proxyStdout?: boolean; // print internal stdout to console @@ -95,10 +97,9 @@ function createPool(script: string, options: PoolOptions = {}, logger: Logger) { const allWorkers: Record = {}; // cgroup v2 leaf per child (keyed by pid), when memory enforcement is enabled - const cgroupParent = options.cgroupParent || DEFAULT_CGROUP_PARENT; + const cgroupParent = options.cgroupParent || resolveSelfCgroup(); const cgroupEnabled = - !!options.cgroupMemoryLimitMb && - isCgroupV2Available(cgroupParent, logger); + !!options.cgroupMemoryLimitMb && isCgroupV2Available(cgroupParent, logger); const cgroups: Record = {}; // Tear down a child's leaf cgroup once it has been killed (best-effort). @@ -141,7 +142,7 @@ function createPool(script: string, options: PoolOptions = {}, logger: Logger) { // lives for the lifetime of the (reused) child and is removed when it dies. if (cgroupEnabled && child.pid) { cgroups[child.pid] = createChildCgroup( - cgroupParent, + cgroupParent!, child.pid, options.cgroupMemoryLimitMb! * 1024 * 1024, logger diff --git a/packages/engine-multi/test/worker/cgroup-enforcement.test.ts b/packages/engine-multi/test/worker/cgroup-enforcement.test.ts index 7054097c2..bd696a1f7 100644 --- a/packages/engine-multi/test/worker/cgroup-enforcement.test.ts +++ b/packages/engine-multi/test/worker/cgroup-enforcement.test.ts @@ -1,29 +1,38 @@ import test from 'ava'; -import os from 'node:os'; import path from 'node:path'; import { createMockLogger } from '@openfn/logger'; import createPool from '../../src/worker/pool'; +import { + isCgroupV2Available, + resolveSelfCgroup, + _resetAvailabilityCache, +} from '../../src/worker/cgroup'; // These tests prove the kernel actually OOM-kills a runaway run via the -// cgroup's memory.max ceiling. That only works on Linux with a writable -// cgroup v2 hierarchy (e.g. a privileged container), so they are skipped -// everywhere else — including CI on macOS. +// cgroup's memory.max ceiling. That needs a writable, delegated cgroup v2 +// subtree (e.g. a privileged container), so they are skipped everywhere else. +// The gate must be real availability, not just the platform: without a +// working cgroup there is no limit of any kind on blowNativeMemory, and it +// would happily eat the whole host. // // docker run --rm -it --privileged -v "$PWD":/kit -w /kit node:24 bash // corepack enable && pnpm install && pnpm --filter @openfn/engine-multi build // cd packages/engine-multi && pnpm ava test/worker/cgroup-enforcement.test.ts -const linuxOnly = os.platform() === 'linux' ? test.serial : test.serial.skip; +const logger = createMockLogger(); +const available = isCgroupV2Available(resolveSelfCgroup(), logger); +_resetAvailabilityCache(); + +const cgroupTest = available ? test.serial : test.serial.skip; const workerPath = path.resolve('dist/test/worker-functions.js'); -const logger = createMockLogger(); // Ceiling above Node's baseline RSS but low enough that blowNativeMemory // crosses it almost immediately. memoryLimitMb is deliberately left unset so // the V8 heap limit can't be what kills the run — only the cgroup can. const cgroupMemoryLimitMb = 200; -linuxOnly( +cgroupTest( 'cgroup OOM-kills a run that exceeds memory.max and surfaces OOMError', async (t) => { const pool = createPool(workerPath, { cgroupMemoryLimitMb }, logger); @@ -37,7 +46,7 @@ linuxOnly( } ); -linuxOnly('pool recovers after a cgroup OOM kill', async (t) => { +cgroupTest('pool recovers after a cgroup OOM kill', async (t) => { const pool = createPool(workerPath, { cgroupMemoryLimitMb }, logger); await t.throwsAsync(() => pool.exec('blowNativeMemory', []), { diff --git a/packages/engine-multi/test/worker/cgroup.test.ts b/packages/engine-multi/test/worker/cgroup.test.ts index 28a779338..eab90ed04 100644 --- a/packages/engine-multi/test/worker/cgroup.test.ts +++ b/packages/engine-multi/test/worker/cgroup.test.ts @@ -3,11 +3,12 @@ import os from 'node:os'; import { createMockLogger } from '@openfn/logger'; import { - DEFAULT_CGROUP_PARENT, createChildCgroup, hasOomKill, isCgroupV2Available, + parseProcSelfCgroup, removeChildCgroup, + resolveSelfCgroup, _resetAvailabilityCache, } from '../../src/worker/cgroup'; @@ -19,27 +20,71 @@ test.beforeEach(() => { _resetAvailabilityCache(); }); -test('DEFAULT_CGROUP_PARENT lives under the cgroup root', (t) => { - t.is(DEFAULT_CGROUP_PARENT, '/sys/fs/cgroup/openfn'); +test('parseProcSelfCgroup resolves the v2 entry against the cgroup mount', (t) => { + t.is( + parseProcSelfCgroup('0::/system.slice/openfn.service\n'), + '/sys/fs/cgroup/system.slice/openfn.service' + ); +}); + +test('parseProcSelfCgroup handles the namespace root', (t) => { + t.is(parseProcSelfCgroup('0::/\n'), '/sys/fs/cgroup'); +}); + +test('parseProcSelfCgroup returns null for a v1-only listing', (t) => { + const v1 = ['12:memory:/user.slice', '3:cpu,cpuacct:/user.slice', ''].join( + '\n' + ); + t.is(parseProcSelfCgroup(v1), null); }); -// cgroups don't exist on macOS/Windows, so the whole module must no-op there. if (!isLinux) { - test('isCgroupV2Available returns false on non-linux hosts', (t) => { - t.false(isCgroupV2Available(DEFAULT_CGROUP_PARENT, logger)); + test('resolveSelfCgroup returns null on non-linux hosts', (t) => { + t.is(resolveSelfCgroup(), null); }); - test('availability probe is cached and only warns once', (t) => { - const l = createMockLogger('test', { level: 'debug' }); - isCgroupV2Available('/sys/fs/cgroup/test-cache', l); - isCgroupV2Available('/sys/fs/cgroup/test-cache', l); - isCgroupV2Available('/sys/fs/cgroup/test-cache', l); + test('isCgroupV2Available returns false on non-linux hosts', (t) => { + t.false(isCgroupV2Available('/sys/fs/cgroup/test', logger)); + }); +} - const warnings = l._history.filter((h: any) => h[0] === 'warn'); - t.is(warnings.length, 1); +if (isLinux) { + test('resolveSelfCgroup returns a path under the cgroup mount', (t) => { + const self = resolveSelfCgroup(); + // null is legitimate on a cgroup v1 host; otherwise the path must be + // inside the mount and never a leader leaf (the enclosing cgroup is + // returned instead) + if (self !== null) { + t.true(self.startsWith('/sys/fs/cgroup')); + t.not(self.split('/').pop(), 'leader'); + } else { + t.pass(); + } }); } +test('isCgroupV2Available returns false for a null parent', (t) => { + t.false(isCgroupV2Available(null, logger)); +}); + +test('isCgroupV2Available returns false for a parent outside the cgroup mount', (t) => { + t.false(isCgroupV2Available('/tmp/not-a-cgroup', logger)); +}); + +test('isCgroupV2Available returns false for a missing parent', (t) => { + t.false(isCgroupV2Available('/sys/fs/cgroup/this/does/not/exist', logger)); +}); + +test('availability probe is cached and only warns once', (t) => { + const l = createMockLogger('test', { level: 'debug' }); + isCgroupV2Available('/sys/fs/cgroup/test-cache', l); + isCgroupV2Available('/sys/fs/cgroup/test-cache', l); + isCgroupV2Available('/sys/fs/cgroup/test-cache', l); + + const warnings = l._history.filter((h: any) => h[0] === 'warn'); + t.is(warnings.length, 1); +}); + test('createChildCgroup returns null when the parent is not writable', (t) => { const handle = createChildCgroup( '/this/path/does/not/exist', diff --git a/packages/ws-worker/src/util/cli.ts b/packages/ws-worker/src/util/cli.ts index ca3bc9e14..caa722a68 100644 --- a/packages/ws-worker/src/util/cli.ts +++ b/packages/ws-worker/src/util/cli.ts @@ -242,7 +242,7 @@ export default function parseArgs(argv: string[]): Args { }) .option('cgroup-parent', { description: - 'Parent cgroup path under which per-run leaf cgroups are created. Must be writable (cgroup delegation/root). Default /sys/fs/cgroup/openfn. Env: WORKER_CGROUP_PARENT', + 'Parent cgroup path under which per-run leaf cgroups are created. Defaults to the cgroup the worker was started in; overriding generally requires root. Env: WORKER_CGROUP_PARENT', type: 'string', }) From c22d6a2cddc5770a546fda2604439ca0eefebaaf Mon Sep 17 00:00:00 2001 From: Joe Clark Date: Fri, 17 Jul 2026 12:25:48 +0100 Subject: [PATCH 06/15] remove seperate cgroup memory option: instead derive from runmemorymb --- packages/engine-multi/src/api.ts | 1 - packages/engine-multi/src/api/call-worker.ts | 3 --- packages/engine-multi/src/engine.ts | 2 -- packages/engine-multi/src/worker/pool.ts | 15 +++++++++------ .../test/worker/cgroup-enforcement.test.ts | 15 +++++++++------ packages/ws-worker/src/start.ts | 3 --- packages/ws-worker/src/util/cli.ts | 10 ---------- 7 files changed, 18 insertions(+), 31 deletions(-) diff --git a/packages/engine-multi/src/api.ts b/packages/engine-multi/src/api.ts index 783ac47d0..43926991b 100644 --- a/packages/engine-multi/src/api.ts +++ b/packages/engine-multi/src/api.ts @@ -58,7 +58,6 @@ const createAPI = async function ( maxWorkers: options.maxWorkers, memoryLimitMb: options.memoryLimitMb || DEFAULT_MEMORY_LIMIT, - cgroupMemoryLimitMb: options.cgroupMemoryLimitMb, cgroupParent: options.cgroupParent, runTimeoutMs: options.runTimeoutMs, diff --git a/packages/engine-multi/src/api/call-worker.ts b/packages/engine-multi/src/api/call-worker.ts index 8a8cbdea5..748f4c6c4 100644 --- a/packages/engine-multi/src/api/call-worker.ts +++ b/packages/engine-multi/src/api/call-worker.ts @@ -13,7 +13,6 @@ 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 }; @@ -29,7 +28,6 @@ export default function initWorkers( env = {}, maxWorkers = 5, memoryLimitMb, - cgroupMemoryLimitMb, cgroupParent, proxyStdout = false, } = options; @@ -40,7 +38,6 @@ export default function initWorkers( maxWorkers, env, memoryLimitMb, - cgroupMemoryLimitMb, cgroupParent, proxyStdout, }, diff --git a/packages/engine-multi/src/engine.ts b/packages/engine-multi/src/engine.ts index 2dc5ce545..eb05b83e8 100644 --- a/packages/engine-multi/src/engine.ts +++ b/packages/engine-multi/src/engine.ts @@ -75,7 +75,6 @@ export type EngineOptions = { maxWorkers?: number; memoryLimitMb?: number; stateLimitMb?: number; - cgroupMemoryLimitMb?: number; cgroupParent?: string; payloadLimitMb?: number; logPayloadLimitMb?: number; @@ -144,7 +143,6 @@ const createEngine = async ( { maxWorkers: options.maxWorkers, memoryLimitMb: defaultMemoryLimit, - cgroupMemoryLimitMb: options.cgroupMemoryLimitMb, cgroupParent: options.cgroupParent, proxyStdout: options.proxyStdout, }, diff --git a/packages/engine-multi/src/worker/pool.ts b/packages/engine-multi/src/worker/pool.ts index 199d2c920..61e9d600d 100644 --- a/packages/engine-multi/src/worker/pool.ts +++ b/packages/engine-multi/src/worker/pool.ts @@ -22,15 +22,15 @@ import { resolveSelfCgroup, } from './cgroup'; +// Constant memory overhead to apply to the cgroup ceiling +const CGROUP_MEMORY_OVERHEAD_MB = 128; + export type PoolOptions = { capacity?: number; // defaults to 5 maxWorkers?: number; // alias for capacity. Which is best? env?: Record; // default environment for workers memoryLimitMb?: number; // --max-old-space-size for child processes - // Hard memory.max ceiling (mb) applied to each child via a cgroup v2 leaf. - // Best-effort: ignored on hosts without a writable cgroup v2 hierarchy. - cgroupMemoryLimitMb?: number; // Parent cgroup under which per-child leaf cgroups are created. Defaults to // the cgroup this process was started in; overriding it generally requires // root (the kernel checks the common ancestor's cgroup.procs on migration) @@ -98,8 +98,11 @@ function createPool(script: string, options: PoolOptions = {}, logger: Logger) { // cgroup v2 leaf per child (keyed by pid), when memory enforcement is enabled const cgroupParent = options.cgroupParent || resolveSelfCgroup(); + const cgroupMemoryLimitMb = options.memoryLimitMb + ? options.memoryLimitMb + CGROUP_MEMORY_OVERHEAD_MB + : undefined; const cgroupEnabled = - !!options.cgroupMemoryLimitMb && isCgroupV2Available(cgroupParent, logger); + !!cgroupMemoryLimitMb && isCgroupV2Available(cgroupParent, logger); const cgroups: Record = {}; // Tear down a child's leaf cgroup once it has been killed (best-effort). @@ -144,7 +147,7 @@ function createPool(script: string, options: PoolOptions = {}, logger: Logger) { cgroups[child.pid] = createChildCgroup( cgroupParent!, child.pid, - options.cgroupMemoryLimitMb! * 1024 * 1024, + cgroupMemoryLimitMb! * 1024 * 1024, logger ); } @@ -200,7 +203,7 @@ function createPool(script: string, options: PoolOptions = {}, logger: Logger) { if (handle && hasOomKill(handle)) { logger.error( `pool: worker ${worker.pid} was killed by the OS for exceeding ` + - `its cgroup memory limit (${options.cgroupMemoryLimitMb}mb)` + `its cgroup memory limit (${cgroupMemoryLimitMb}mb)` ); killWorker(worker); // restore a placeholder to the queue diff --git a/packages/engine-multi/test/worker/cgroup-enforcement.test.ts b/packages/engine-multi/test/worker/cgroup-enforcement.test.ts index bd696a1f7..879224efb 100644 --- a/packages/engine-multi/test/worker/cgroup-enforcement.test.ts +++ b/packages/engine-multi/test/worker/cgroup-enforcement.test.ts @@ -27,15 +27,18 @@ const cgroupTest = available ? test.serial : test.serial.skip; const workerPath = path.resolve('dist/test/worker-functions.js'); -// Ceiling above Node's baseline RSS but low enough that blowNativeMemory -// crosses it almost immediately. memoryLimitMb is deliberately left unset so -// the V8 heap limit can't be what kills the run — only the cgroup can. -const cgroupMemoryLimitMb = 200; +// The cgroup ceiling is now derived as memoryLimitMb + a fixed headroom (see +// CGROUP_MEMORY_HEADROOM_MB in pool.ts), so this sets memoryLimitMb low +// enough that the resulting ceiling is above Node's baseline RSS but low +// enough that blowNativeMemory crosses it almost immediately. blowNativeMemory +// allocates native (off V8 heap) memory, so --max-old-space-size can't be +// what kills the run here — only the cgroup can. +const memoryLimitMb = 72; // + 128mb headroom = 200mb effective ceiling cgroupTest( 'cgroup OOM-kills a run that exceeds memory.max and surfaces OOMError', async (t) => { - const pool = createPool(workerPath, { cgroupMemoryLimitMb }, logger); + const pool = createPool(workerPath, { memoryLimitMb }, logger); const err = await t.throwsAsync(() => pool.exec('blowNativeMemory', []), { name: 'OOMError', @@ -47,7 +50,7 @@ cgroupTest( ); cgroupTest('pool recovers after a cgroup OOM kill', async (t) => { - const pool = createPool(workerPath, { cgroupMemoryLimitMb }, logger); + const pool = createPool(workerPath, { memoryLimitMb }, logger); await t.throwsAsync(() => pool.exec('blowNativeMemory', []), { name: 'OOMError', diff --git a/packages/ws-worker/src/start.ts b/packages/ws-worker/src/start.ts index f37f8b2f1..cd5669ace 100644 --- a/packages/ws-worker/src/start.ts +++ b/packages/ws-worker/src/start.ts @@ -112,9 +112,6 @@ if (args.mock) { repoDir: args.repoDir, memoryLimitMb: args.runMemory, stateLimitMb: args.stateMemory, - // Hard cgroup ceiling sits above the V8 heap limit so GC fires first; - // default to run-memory + 128mb of headroom for native/buffer allocations. - cgroupMemoryLimitMb: args.cgroupMemory ?? (args.runMemory ?? 500) + 128, cgroupParent: args.cgroupParent, maxWorkers: effectiveCapacity, statePropsToRemove: args.statePropsToRemove, diff --git a/packages/ws-worker/src/util/cli.ts b/packages/ws-worker/src/util/cli.ts index caa722a68..3799043fd 100644 --- a/packages/ws-worker/src/util/cli.ts +++ b/packages/ws-worker/src/util/cli.ts @@ -16,7 +16,6 @@ type Args = { batchLimit?: number; batchLogs: boolean; capacity: number; - cgroupMemory?: number; cgroupParent?: string; workloops?: string; claimTimeoutSeconds?: number; @@ -84,7 +83,6 @@ export default function parseArgs(argv: string[]): Args { WORKER_BATCH_LIMIT, WORKER_BATCH_LOGS, WORKER_CAPACITY, - WORKER_CGROUP_MEMORY_MB, WORKER_CGROUP_PARENT, WORKER_CLAIM_TIMEOUT_SECONDS, WORKER_COLLECTIONS_URL, @@ -235,11 +233,6 @@ export default function parseArgs(argv: string[]): Args { 'Maximum memory allocated to a single run, in mb. Env: WORKER_MAX_PAYLOAD_MB', type: 'number', }) - .option('cgroup-memory', { - description: - 'Hard memory ceiling (mb) enforced per run via a cgroup v2 leaf. Linux only; falls back to heap-limit only when unavailable. Defaults to run-memory + 128. Set to 0 to disable. Env: WORKER_CGROUP_MEMORY_MB', - type: 'number', - }) .option('cgroup-parent', { description: 'Parent cgroup path under which per-run leaf cgroups are created. Defaults to the cgroup the worker was started in; overriding generally requires root. Env: WORKER_CGROUP_PARENT', @@ -353,9 +346,6 @@ export default function parseArgs(argv: string[]): Args { (WORKER_MAX_STATE_MEMORY_MB ? parseInt(WORKER_MAX_STATE_MEMORY_MB, 10) : undefined), - cgroupMemory: - args.cgroupMemory ?? - (WORKER_CGROUP_MEMORY_MB ? parseInt(WORKER_CGROUP_MEMORY_MB) : undefined), cgroupParent: setArg(args.cgroupParent, WORKER_CGROUP_PARENT) as | string | undefined, From a762a1053a67d652f965b1183970a1f6f4c8919b Mon Sep 17 00:00:00 2001 From: Joe Clark Date: Fri, 17 Jul 2026 13:39:44 +0100 Subject: [PATCH 07/15] remove cgroup parent name --- packages/engine-multi/src/api.ts | 1 - packages/engine-multi/src/api/call-worker.ts | 3 --- packages/engine-multi/src/engine.ts | 2 -- packages/engine-multi/src/worker/pool.ts | 7 +------ packages/ws-worker/src/start.ts | 1 - packages/ws-worker/src/util/cli.ts | 11 ----------- 6 files changed, 1 insertion(+), 24 deletions(-) diff --git a/packages/engine-multi/src/api.ts b/packages/engine-multi/src/api.ts index 43926991b..29450b16a 100644 --- a/packages/engine-multi/src/api.ts +++ b/packages/engine-multi/src/api.ts @@ -58,7 +58,6 @@ const createAPI = async function ( maxWorkers: options.maxWorkers, memoryLimitMb: options.memoryLimitMb || DEFAULT_MEMORY_LIMIT, - cgroupParent: options.cgroupParent, runTimeoutMs: options.runTimeoutMs, statePropsToRemove: options.statePropsToRemove ?? [ diff --git a/packages/engine-multi/src/api/call-worker.ts b/packages/engine-multi/src/api/call-worker.ts index 748f4c6c4..c3a062407 100644 --- a/packages/engine-multi/src/api/call-worker.ts +++ b/packages/engine-multi/src/api/call-worker.ts @@ -13,7 +13,6 @@ type WorkerOptions = { env?: any; timeout?: number; // ms memoryLimitMb?: number; - cgroupParent?: string; // parent cgroup for per-child leaves proxyStdout?: boolean; // print internal stdout to console }; @@ -28,7 +27,6 @@ export default function initWorkers( env = {}, maxWorkers = 5, memoryLimitMb, - cgroupParent, proxyStdout = false, } = options; @@ -38,7 +36,6 @@ export default function initWorkers( maxWorkers, env, memoryLimitMb, - cgroupParent, proxyStdout, }, logger diff --git a/packages/engine-multi/src/engine.ts b/packages/engine-multi/src/engine.ts index eb05b83e8..38467b97c 100644 --- a/packages/engine-multi/src/engine.ts +++ b/packages/engine-multi/src/engine.ts @@ -75,7 +75,6 @@ export type EngineOptions = { maxWorkers?: number; memoryLimitMb?: number; stateLimitMb?: number; - cgroupParent?: string; payloadLimitMb?: number; logPayloadLimitMb?: number; repoDir: string; @@ -143,7 +142,6 @@ const createEngine = async ( { maxWorkers: options.maxWorkers, memoryLimitMb: defaultMemoryLimit, - cgroupParent: options.cgroupParent, proxyStdout: options.proxyStdout, }, options.logger diff --git a/packages/engine-multi/src/worker/pool.ts b/packages/engine-multi/src/worker/pool.ts index 61e9d600d..0d16fd105 100644 --- a/packages/engine-multi/src/worker/pool.ts +++ b/packages/engine-multi/src/worker/pool.ts @@ -31,11 +31,6 @@ export type PoolOptions = { env?: Record; // default environment for workers memoryLimitMb?: number; // --max-old-space-size for child processes - // Parent cgroup under which per-child leaf cgroups are created. Defaults to - // the cgroup this process was started in; overriding it generally requires - // root (the kernel checks the common ancestor's cgroup.procs on migration) - cgroupParent?: string; - proxyStdout?: boolean; // print internal stdout to console }; @@ -97,7 +92,7 @@ function createPool(script: string, options: PoolOptions = {}, logger: Logger) { const allWorkers: Record = {}; // cgroup v2 leaf per child (keyed by pid), when memory enforcement is enabled - const cgroupParent = options.cgroupParent || resolveSelfCgroup(); + const cgroupParent = resolveSelfCgroup(); const cgroupMemoryLimitMb = options.memoryLimitMb ? options.memoryLimitMb + CGROUP_MEMORY_OVERHEAD_MB : undefined; diff --git a/packages/ws-worker/src/start.ts b/packages/ws-worker/src/start.ts index cd5669ace..4e08e2601 100644 --- a/packages/ws-worker/src/start.ts +++ b/packages/ws-worker/src/start.ts @@ -112,7 +112,6 @@ if (args.mock) { repoDir: args.repoDir, memoryLimitMb: args.runMemory, stateLimitMb: args.stateMemory, - cgroupParent: args.cgroupParent, maxWorkers: effectiveCapacity, statePropsToRemove: args.statePropsToRemove, runTimeoutMs: args.maxRunDurationSeconds * 1000, diff --git a/packages/ws-worker/src/util/cli.ts b/packages/ws-worker/src/util/cli.ts index 3799043fd..8114fa34e 100644 --- a/packages/ws-worker/src/util/cli.ts +++ b/packages/ws-worker/src/util/cli.ts @@ -16,7 +16,6 @@ type Args = { batchLimit?: number; batchLogs: boolean; capacity: number; - cgroupParent?: string; workloops?: string; claimTimeoutSeconds?: number; collectionsUrl?: string; @@ -83,7 +82,6 @@ export default function parseArgs(argv: string[]): Args { WORKER_BATCH_LIMIT, WORKER_BATCH_LOGS, WORKER_CAPACITY, - WORKER_CGROUP_PARENT, WORKER_CLAIM_TIMEOUT_SECONDS, WORKER_COLLECTIONS_URL, WORKER_COLLECTIONS_VERSION, @@ -233,12 +231,6 @@ export default function parseArgs(argv: string[]): Args { 'Maximum memory allocated to a single run, in mb. Env: WORKER_MAX_PAYLOAD_MB', type: 'number', }) - .option('cgroup-parent', { - description: - 'Parent cgroup path under which per-run leaf cgroups are created. Defaults to the cgroup the worker was started in; overriding generally requires root. Env: WORKER_CGROUP_PARENT', - type: 'string', - }) - .option('max-run-duration-seconds', { alias: 't', description: @@ -346,9 +338,6 @@ export default function parseArgs(argv: string[]): Args { (WORKER_MAX_STATE_MEMORY_MB ? parseInt(WORKER_MAX_STATE_MEMORY_MB, 10) : undefined), - cgroupParent: setArg(args.cgroupParent, WORKER_CGROUP_PARENT) as - | string - | undefined, payloadMemory: setArg(args.payloadMemory, WORKER_MAX_PAYLOAD_MB, 10), logPayloadMemory: setArg( args.logPayloadMemory, From 5c9603f060804bc62a8af81b5f09ec545335ab60 Mon Sep 17 00:00:00 2001 From: Joe Clark Date: Fri, 17 Jul 2026 13:48:26 +0100 Subject: [PATCH 08/15] fix enforcement tests --- packages/engine-multi/ava.config.cjs | 8 +++++++- packages/engine-multi/package.json | 1 + 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/packages/engine-multi/ava.config.cjs b/packages/engine-multi/ava.config.cjs index 21d373937..842ebd7d6 100644 --- a/packages/engine-multi/ava.config.cjs +++ b/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', + ], }; diff --git a/packages/engine-multi/package.json b/packages/engine-multi/package.json index 35ef12543..e38df3112 100644 --- a/packages/engine-multi/package.json +++ b/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", From 9da4fe4c71b289b07fea844c55afc394985657cf Mon Sep 17 00:00:00 2001 From: Joe Clark Date: Fri, 17 Jul 2026 13:58:30 +0100 Subject: [PATCH 09/15] simplify cgroup tests --- packages/engine-multi/src/worker/pool.ts | 28 +++++++++--- .../test/worker/cgroup-enforcement.test.ts | 43 +++++++++++-------- 2 files changed, 45 insertions(+), 26 deletions(-) diff --git a/packages/engine-multi/src/worker/pool.ts b/packages/engine-multi/src/worker/pool.ts index 0d16fd105..28fb2546d 100644 --- a/packages/engine-multi/src/worker/pool.ts +++ b/packages/engine-multi/src/worker/pool.ts @@ -25,11 +25,22 @@ import { // Constant memory overhead to apply to the cgroup ceiling const CGROUP_MEMORY_OVERHEAD_MB = 128; +export type MemoryEnforcement = { + // Sets node's --max-old-space-size on child processes + // Default true + oldspace?: boolean; + + // Uses cgroups to set a hard ceiling on child processes through the kernel + // Default false. + cgroup?: boolean; +}; + export type PoolOptions = { capacity?: number; // defaults to 5 maxWorkers?: number; // alias for capacity. Which is best? env?: Record; // default environment for workers - memoryLimitMb?: number; // --max-old-space-size for child processes + memoryLimitMb?: number; // Set the maximum runtime memory a child process can consume + memoryEnforcement?: MemoryEnforcement; proxyStdout?: boolean; // print internal stdout to console }; @@ -91,13 +102,16 @@ function createPool(script: string, options: PoolOptions = {}, logger: Logger) { // Keep track of all the workers we created const allWorkers: Record = {}; - // cgroup v2 leaf per child (keyed by pid), when memory enforcement is enabled + const enforceOldspace = options.memoryEnforcement?.oldspace ?? true; + + // cgroup v2 leaf per child (keyed by pid), when cgroup memory enforcement is enabled const cgroupParent = resolveSelfCgroup(); - const cgroupMemoryLimitMb = options.memoryLimitMb - ? options.memoryLimitMb + CGROUP_MEMORY_OVERHEAD_MB - : undefined; + const cgroupMemoryLimitMb = + options.memoryLimitMb && options.memoryLimitMb + CGROUP_MEMORY_OVERHEAD_MB; const cgroupEnabled = - !!cgroupMemoryLimitMb && isCgroupV2Available(cgroupParent, logger); + options.memoryEnforcement?.cgroup && + !!cgroupMemoryLimitMb && + isCgroupV2Available(cgroupParent, logger); const cgroups: Record = {}; // Tear down a child's leaf cgroup once it has been killed (best-effort). @@ -113,7 +127,7 @@ function createPool(script: string, options: PoolOptions = {}, logger: Logger) { if (!maybeChild) { // create a new child process and load the module script into it const execArgv = ['--experimental-vm-modules', '--no-warnings']; - if (options.memoryLimitMb) { + if (enforceOldspace && options.memoryLimitMb) { execArgv.push(`--max-old-space-size=${options.memoryLimitMb}`); } diff --git a/packages/engine-multi/test/worker/cgroup-enforcement.test.ts b/packages/engine-multi/test/worker/cgroup-enforcement.test.ts index 879224efb..beb53028e 100644 --- a/packages/engine-multi/test/worker/cgroup-enforcement.test.ts +++ b/packages/engine-multi/test/worker/cgroup-enforcement.test.ts @@ -1,3 +1,12 @@ +/** + * These tests prove the kernel actually OOM-kills a runaway run via the cgroup's memory.max ceiling. + * Tests run with node's oldspace enforcement disabled: cgroups are the only lever we can use to trigger an OOM error. + * + * !!!IMPORTANT!! + * + * This is NOT safe to run directly from a terminal as the owning cgroup will be terminated + * Use `pnpm test:cgroup` to run the test in a detatched process + */ import test from 'ava'; import path from 'node:path'; import { createMockLogger } from '@openfn/logger'; @@ -9,17 +18,8 @@ import { _resetAvailabilityCache, } from '../../src/worker/cgroup'; -// These tests prove the kernel actually OOM-kills a runaway run via the -// cgroup's memory.max ceiling. That needs a writable, delegated cgroup v2 -// subtree (e.g. a privileged container), so they are skipped everywhere else. -// The gate must be real availability, not just the platform: without a -// working cgroup there is no limit of any kind on blowNativeMemory, and it -// would happily eat the whole host. -// -// docker run --rm -it --privileged -v "$PWD":/kit -w /kit node:24 bash -// corepack enable && pnpm install && pnpm --filter @openfn/engine-multi build -// cd packages/engine-multi && pnpm ava test/worker/cgroup-enforcement.test.ts const logger = createMockLogger(); + const available = isCgroupV2Available(resolveSelfCgroup(), logger); _resetAvailabilityCache(); @@ -27,18 +27,19 @@ const cgroupTest = available ? test.serial : test.serial.skip; const workerPath = path.resolve('dist/test/worker-functions.js'); -// The cgroup ceiling is now derived as memoryLimitMb + a fixed headroom (see -// CGROUP_MEMORY_HEADROOM_MB in pool.ts), so this sets memoryLimitMb low -// enough that the resulting ceiling is above Node's baseline RSS but low -// enough that blowNativeMemory crosses it almost immediately. blowNativeMemory -// allocates native (off V8 heap) memory, so --max-old-space-size can't be -// what kills the run here — only the cgroup can. -const memoryLimitMb = 72; // + 128mb headroom = 200mb effective ceiling +const memoryLimitMb = 72; // + 128mb headroom = 200mb effective cgroup ceiling + +// Disable node's oldspace memory limit so that croup is the only variable in play +const memoryEnforcement = { cgroup: true, oldspace: false }; cgroupTest( 'cgroup OOM-kills a run that exceeds memory.max and surfaces OOMError', async (t) => { - const pool = createPool(workerPath, { memoryLimitMb }, logger); + const pool = createPool( + workerPath, + { memoryLimitMb, memoryEnforcement }, + logger + ); const err = await t.throwsAsync(() => pool.exec('blowNativeMemory', []), { name: 'OOMError', @@ -50,7 +51,11 @@ cgroupTest( ); cgroupTest('pool recovers after a cgroup OOM kill', async (t) => { - const pool = createPool(workerPath, { memoryLimitMb }, logger); + const pool = createPool( + workerPath, + { memoryLimitMb, memoryEnforcement }, + logger + ); await t.throwsAsync(() => pool.exec('blowNativeMemory', []), { name: 'OOMError', From c62219aa37ce60dcf6e6adbe9ed997011a6d375f Mon Sep 17 00:00:00 2001 From: Joe Clark Date: Fri, 17 Jul 2026 14:07:29 +0100 Subject: [PATCH 10/15] add options to feed --cgroup flag into engine --- packages/engine-multi/src/api.ts | 1 + packages/engine-multi/src/api/call-worker.ts | 5 ++++- packages/engine-multi/src/engine.ts | 3 +++ packages/ws-worker/src/start.ts | 1 + packages/ws-worker/src/util/cli.ts | 8 ++++++++ 5 files changed, 17 insertions(+), 1 deletion(-) diff --git a/packages/engine-multi/src/api.ts b/packages/engine-multi/src/api.ts index 29450b16a..593196b62 100644 --- a/packages/engine-multi/src/api.ts +++ b/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 ?? [ diff --git a/packages/engine-multi/src/api/call-worker.ts b/packages/engine-multi/src/api/call-worker.ts index c3a062407..d2f648b0c 100644 --- a/packages/engine-multi/src/api/call-worker.ts +++ b/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 diff --git a/packages/engine-multi/src/engine.ts b/packages/engine-multi/src/engine.ts index 38467b97c..c66c1d388 100644 --- a/packages/engine-multi/src/engine.ts +++ b/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 diff --git a/packages/ws-worker/src/start.ts b/packages/ws-worker/src/start.ts index 4e08e2601..e8650ee72 100644 --- a/packages/ws-worker/src/start.ts +++ b/packages/ws-worker/src/start.ts @@ -111,6 +111,7 @@ if (args.mock) { const engineOptions = { repoDir: args.repoDir, memoryLimitMb: args.runMemory, + memoryEnforcement: { cgroup: args.cgroup }, stateLimitMb: args.stateMemory, maxWorkers: effectiveCapacity, statePropsToRemove: args.statePropsToRemove, diff --git a/packages/ws-worker/src/util/cli.ts b/packages/ws-worker/src/util/cli.ts index 8114fa34e..cde590e50 100644 --- a/packages/ws-worker/src/util/cli.ts +++ b/packages/ws-worker/src/util/cli.ts @@ -16,6 +16,7 @@ type Args = { batchLimit?: number; batchLogs: boolean; capacity: number; + cgroup?: boolean; workloops?: string; claimTimeoutSeconds?: number; collectionsUrl?: string; @@ -82,6 +83,7 @@ export default function parseArgs(argv: string[]): Args { WORKER_BATCH_LIMIT, WORKER_BATCH_LOGS, WORKER_CAPACITY, + WORKER_CGROUP, WORKER_CLAIM_TIMEOUT_SECONDS, WORKER_COLLECTIONS_URL, WORKER_COLLECTIONS_VERSION, @@ -231,6 +233,11 @@ export default function parseArgs(argv: string[]): Args { 'Maximum memory allocated to a single run, in mb. Env: WORKER_MAX_PAYLOAD_MB', type: 'number', }) + .option('cgroup', { + description: + 'Enforce run-memory as a hard ceiling via a cgroup v2 leaf, in addition to --max-old-space-size. Linux only; falls back to heap-limit only if a writable, delegated cgroup is unavailable. Default false (disabled entirely, no availability check). Env: WORKER_CGROUP', + type: 'boolean', + }) .option('max-run-duration-seconds', { alias: 't', description: @@ -326,6 +333,7 @@ export default function parseArgs(argv: string[]): Args { log: setArg(args.log, WORKER_LOG_LEVEL as LogLevel, 'debug'), backoff: setArg(args.backoff, WORKER_BACKOFF, '1/10'), capacity: setArg(args.capacity, WORKER_CAPACITY, DEFAULT_WORKER_CAPACITY), + cgroup: setArg(args.cgroup, WORKER_CGROUP, false), statePropsToRemove: setArg( args.statePropsToRemove, WORKER_STATE_PROPS_TO_REMOVE, From 20bc595dc2bafc810ba370dd2547a488be57bbdb Mon Sep 17 00:00:00 2001 From: Joe Clark Date: Fri, 17 Jul 2026 14:08:33 +0100 Subject: [PATCH 11/15] in docker image, enable cgroups --- Dockerfile | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Dockerfile b/Dockerfile index 72805704c..fb1cd447b 100644 --- a/Dockerfile +++ b/Dockerfile @@ -25,4 +25,4 @@ WORKDIR /app/packages/ws-worker # ------------------------------------------------------------------------------ EXPOSE 2222 -CMD [ "node", "./dist/start.js"] \ No newline at end of file +CMD [ "node", "./dist/start.js" "--cgroups"] \ No newline at end of file From 3ce1fc1ab2c620778bb93e60d935ca61bcb8b169 Mon Sep 17 00:00:00 2001 From: Joe Clark Date: Fri, 17 Jul 2026 14:17:21 +0100 Subject: [PATCH 12/15] tweaks --- Dockerfile | 2 +- packages/engine-multi/README.md | 14 +++++++------- packages/engine-multi/src/worker/cgroup.ts | 7 ++----- packages/ws-worker/src/util/cli.ts | 7 ++++--- 4 files changed, 14 insertions(+), 16 deletions(-) diff --git a/Dockerfile b/Dockerfile index fb1cd447b..72805704c 100644 --- a/Dockerfile +++ b/Dockerfile @@ -25,4 +25,4 @@ WORKDIR /app/packages/ws-worker # ------------------------------------------------------------------------------ EXPOSE 2222 -CMD [ "node", "./dist/start.js" "--cgroups"] \ No newline at end of file +CMD [ "node", "./dist/start.js"] \ No newline at end of file diff --git a/packages/engine-multi/README.md b/packages/engine-multi/README.md index 8ecd33bdd..c6f1fe52e 100644 --- a/packages/engine-multi/README.md +++ b/packages/engine-multi/README.md @@ -97,18 +97,18 @@ An OOMError carries a `source` property (`'heap'` or `'cgroup'`) saying which li 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 | +| 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. +The ws-worker enables cgroup enforcement by default, with `run-memory + 128`mb of headroom for native allocations. Pass `--cgroup-memory 0` (or `WORKER_ENABLE_CGROUP_ENFORCEMENT_MEMORY_MB=0`) to disable it explicitly. ## Module Loader Whitelist diff --git a/packages/engine-multi/src/worker/cgroup.ts b/packages/engine-multi/src/worker/cgroup.ts index 4540e7160..1c4420aa3 100644 --- a/packages/engine-multi/src/worker/cgroup.ts +++ b/packages/engine-multi/src/worker/cgroup.ts @@ -4,11 +4,8 @@ * cgroup level memory enforcement allows us to enforce memory limits at the kernel level, * improving our ability to OOMKill runs. * - * The engine never provisions the cgroup hierarchy itself. The contract is that - * the worker process is *started inside* a delegated, writable cgroup - by a - * systemd unit with Delegate=, by a container runtime, or by a launcher script - * (see docs/cgroup-spec.md). When the contract isn't met, every function here - * degrades to a no-op and the caller falls back to heap-limit-only behaviour. + * cgroup enforcement is disabled by default because the worker needs to run in an appropriately delegated runtime. See readme. + * */ import fs from 'node:fs'; import os from 'node:os'; diff --git a/packages/ws-worker/src/util/cli.ts b/packages/ws-worker/src/util/cli.ts index cde590e50..e6bd9d374 100644 --- a/packages/ws-worker/src/util/cli.ts +++ b/packages/ws-worker/src/util/cli.ts @@ -83,7 +83,7 @@ export default function parseArgs(argv: string[]): Args { WORKER_BATCH_LIMIT, WORKER_BATCH_LOGS, WORKER_CAPACITY, - WORKER_CGROUP, + WORKER_ENABLE_CGROUP_ENFORCEMENT, WORKER_CLAIM_TIMEOUT_SECONDS, WORKER_COLLECTIONS_URL, WORKER_COLLECTIONS_VERSION, @@ -234,8 +234,9 @@ export default function parseArgs(argv: string[]): Args { type: 'number', }) .option('cgroup', { + alias: ['enable-cgroup-enforcement', 'cgroups'], description: - 'Enforce run-memory as a hard ceiling via a cgroup v2 leaf, in addition to --max-old-space-size. Linux only; falls back to heap-limit only if a writable, delegated cgroup is unavailable. Default false (disabled entirely, no availability check). Env: WORKER_CGROUP', + 'Enforce run memory limits applied via cgroups for extra memory hardening. Linux only. Default false. Env: WORKER_ENABLE_CGROUP_ENFORCEMENT', type: 'boolean', }) .option('max-run-duration-seconds', { @@ -333,7 +334,7 @@ export default function parseArgs(argv: string[]): Args { log: setArg(args.log, WORKER_LOG_LEVEL as LogLevel, 'debug'), backoff: setArg(args.backoff, WORKER_BACKOFF, '1/10'), capacity: setArg(args.capacity, WORKER_CAPACITY, DEFAULT_WORKER_CAPACITY), - cgroup: setArg(args.cgroup, WORKER_CGROUP, false), + cgroup: setArg(args.cgroup, WORKER_ENABLE_CGROUP_ENFORCEMENT, false), statePropsToRemove: setArg( args.statePropsToRemove, WORKER_STATE_PROPS_TO_REMOVE, From be3dfc748cd595eb75eadde2eebef2b22dfd7cba Mon Sep 17 00:00:00 2001 From: Joe Clark Date: Fri, 17 Jul 2026 14:32:51 +0100 Subject: [PATCH 13/15] docs --- packages/engine-multi/README.md | 23 ++++++-------------- packages/ws-worker/README.md | 38 +++++++++++++++++++++++++++++++++ 2 files changed, 45 insertions(+), 16 deletions(-) diff --git a/packages/engine-multi/README.md b/packages/engine-multi/README.md index c6f1fe52e..133fc5122 100644 --- a/packages/engine-multi/README.md +++ b/packages/engine-multi/README.md @@ -83,32 +83,23 @@ For a full list of events, see `src/events/ts` (the top-level API events are lis ## Memory Limits -The engine enforces two memory limits on each run: +The engine enforces a memory limits on each run via `memoryLimitMb`. There are two modes of enforcement: -**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. +**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** (`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. +**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 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 | +The engine does not provision the cgroup hierarchy itself. Rather, the host runtime must be started within a writable, cgroup v2 subtree. -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. +The engine creates one leaf per child process within it, applying the memory limit to each. -`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`. +cgroups must be enabled explicitly: the are off by default because unless properly configured, an OOM exception can kill the parenting process group hierarchy. -The ws-worker enables cgroup enforcement by default, with `run-memory + 128`mb of headroom for native allocations. Pass `--cgroup-memory 0` (or `WORKER_ENABLE_CGROUP_ENFORCEMENT_MEMORY_MB=0`) to disable it explicitly. +See Worker documentation for notes about how to start the engine with cgroups enabled. ## Module Loader Whitelist diff --git a/packages/ws-worker/README.md b/packages/ws-worker/README.md index 9a82e8209..c0ca24d50 100644 --- a/packages/ws-worker/README.md +++ b/packages/ws-worker/README.md @@ -51,6 +51,44 @@ Use `-l mock` to connect to a lightning mock server (on the default port). For a list of supported worker and engine options, see src/start.ts +## Enforcing memory limits with cgroups + +Each run's memory limit is enforced by default through node's max-old-space-size, which only constrains heap size. Native and buffer allocations bypass this limit. This can cause the worker to consume more memory than it is technically allowed, which can in turn cause the whole worker process to be killed by its container (ie, kubernetes). + +For tighter enforcement on linux environments, cgroups can be utilised. This approach make the kernel itself enforce the memory size of all run processes. If a single run exceeds its alloted size, that process is terminated and the run is marked as Killed. See engine docs for more details about this. + +To enable cgroup enforcement, pass `--cgroups` or set `WORKER_ENABLE_CGROUP_ENFORCEMENT`. + +### Local Development + +If you try and start a local worker with cgroup enforcement, and a process exceeds its memory, you'll find that the whole owning process is taken down (ie, the hosting terminal). This is because the worker doesn't create its own isolated cgroup — it enforces limits using whatever cgroup it happens to be started in. In an ordinary dev terminal, that cgroup is shared with other things (your shell, other tools, sometimes your whole desktop session), so an OOM kill there can take out more than just the run that went over its limit. + +To safely run with cgroup enforcement, you have to spawn a detatched process and delegate the cgroup: + +``` +systemd-run --user --scope --unit=openfn-worker -p Delegate=yes -- pnpm start +``` + +To close the process press CTRL-C or run: + +``` +systemctl --user stop openfn-worker.scope +``` + +### With Docker + +Cgroup enforcement is off by default even inside a container — you still need to explicitly set `WORKER_ENABLE_CGROUP_ENFORCEMENT` (or pass `--cgroups`) for it to activate. + +Setting the env var isn't enough on its own, though: the container also needs write access to its own cgroup files, which an ordinary `docker run` doesn't grant. Without it, enforcement quietly falls back to heap-limit-only, same as above. + +For local testing, you can grant that access with `--privileged`: + +```bash +docker run --privileged -e WORKER_ENABLE_CGROUP_ENFORCEMENT=true -e WORKER_SECRET=$WORKER_SECRET openfn-worker +``` + +**`--privileged` is not suitable for production.** It grants far more than cgroup write access — full device access, every Linux capability, and disabled seccomp/AppArmor confinement — which is a big escalation for a worker that executes arbitrary job code. For a real deployment, work out the narrowest way to grant cgroup write access with whoever owns your container/orchestration setup, rather than reaching for `--privileged`. + ## Watched Server You can start a dev server (which rebuilds on save) by running: From 6c02d0c6b8ffd6c315743c564ba57c5f40fb72c3 Mon Sep 17 00:00:00 2001 From: Joe Clark Date: Fri, 17 Jul 2026 14:33:14 +0100 Subject: [PATCH 14/15] typo --- packages/ws-worker/README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/ws-worker/README.md b/packages/ws-worker/README.md index c0ca24d50..f9394b844 100644 --- a/packages/ws-worker/README.md +++ b/packages/ws-worker/README.md @@ -63,7 +63,7 @@ To enable cgroup enforcement, pass `--cgroups` or set `WORKER_ENABLE_CGROUP_ENFO If you try and start a local worker with cgroup enforcement, and a process exceeds its memory, you'll find that the whole owning process is taken down (ie, the hosting terminal). This is because the worker doesn't create its own isolated cgroup — it enforces limits using whatever cgroup it happens to be started in. In an ordinary dev terminal, that cgroup is shared with other things (your shell, other tools, sometimes your whole desktop session), so an OOM kill there can take out more than just the run that went over its limit. -To safely run with cgroup enforcement, you have to spawn a detatched process and delegate the cgroup: +To safely run with cgroup enforcement, you have to spawn a detached process and delegate the cgroup: ``` systemd-run --user --scope --unit=openfn-worker -p Delegate=yes -- pnpm start From 783d86b306e94c230edcb865c97f38d26647147a Mon Sep 17 00:00:00 2001 From: Joe Clark Date: Fri, 17 Jul 2026 14:35:09 +0100 Subject: [PATCH 15/15] remove spec --- packages/engine-multi/docs/cgroup-spec.md | 134 ---------------------- 1 file changed, 134 deletions(-) delete mode 100644 packages/engine-multi/docs/cgroup-spec.md diff --git a/packages/engine-multi/docs/cgroup-spec.md b/packages/engine-multi/docs/cgroup-spec.md deleted file mode 100644 index 2bbd3bc0c..000000000 --- a/packages/engine-multi/docs/cgroup-spec.md +++ /dev/null @@ -1,134 +0,0 @@ -# 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::` 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; `/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 - `/leader` leaf — required by the no-internal-processes rule — then - write `+memory` to `/cgroup.subtree_control`. Failure → warn, - fall back. -3. **Per child fork**: `mkdir /run-`; 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.