openvibers/openvibe.host Commits
Serialize host deploys and guard dependency installs by memory (#310)
Changes
@@@ -19,6 +19,12 @@ Production deploys and rollbacks use `ovhost deploy <svc>` and `ovhost rollbackEvery other service keeps the default strategy, `git-checkout`, which is what `ovhost deploy` did before.++## Host deploy admission++++Deploys and rollbacks take one host-wide kernel lock at `/run/lock/openvibe-host-deploy.lock` before preparing a release or changing a checkout. The lock remains held through preflight, switching, restart, readiness and automatic rollback. Another service waits and names the current owner; it fails after the configured lock timeout. A `--wait-idle` operation releases the host lock while it polls streams or recordings, then takes it again before continuing. The existing per-service lock remains held during that wait.++++Before each dependency install, ovhost checks Linux `MemAvailable`. When it is below 1536 MiB, ovhost waits for memory and reports the wait. If memory stays low for 300 seconds, the install does not start and the deploy fails through its normal cleanup path. The root of `host.json` may set `deployGuard.minAvailableMiB`, `deployGuard.memoryWaitSeconds`, `deployGuard.memoryPollSeconds`, and `deployGuard.lockWaitSeconds` (default 1536, 300, 5, and 1800). Each must be a positive integer. This applies to `npm` and `pnpm` installs in both deploy strategies and to dependency reinstall during automatic rollback.++## Deploy wrappersRepository wrappers call `ovhost plan`, `ovhost deploy`, or `ovhost rollback` for their service. If ovhost is missing, the wrapper exits with an error. The `capabilities` command remains available for installed wrappers that still query it.@@@ -364,6 +364,18 @@ const SCENARIOS = [budget: { name: 'calls dropped by a deploy', max: 0 },reason: 'no proof: Chat\'s drain policy is report (a deploy restarts with calls up); a restart ends them with end_reason restart (6e) and people call again',},++ {++ id: '6j', what: 'host-wide deploy locking serializes preparation, while --wait-idle frees the lock', repo: 'host', files: ['test/deploy-guard.test.js'],++ expect: [ok('concurrent deploys serialize preparation across services'), ok('--wait-idle releases the host lock while streams are active')],++ budget: { name: 'overlapping deploy preparations', max: 0 }, measure: asserted(), category: 'resource growth',++ source: 'fake host asserts lock acquisition order and that the second checkout does not move during the first install',++ },++ {++ id: '6k', what: 'low MemAvailable delays dependency install and times out cleanly', repo: 'host', files: ['test/deploy-guard.test.js'],++ expect: [ok('memory guard waits for MemAvailable and then installs'), ok('memory guard times out without starting npm or restarting'), ok('release-layout memory timeout removes the prepared release')],++ budget: { name: 'installs begun below the memory floor', max: 0 }, measure: asserted(), category: 'resource growth',++ source: 'fake host varies MemAvailable and asserts npm and restart are not called on timeout',++ },],},{nnew file mode 100644@@@ -0,0 +1,47 @@++'use strict';++/** Host-wide deploy admission. The executor holds a kernel flock until release(). */++++const LOCK_FILE = '/run/lock/openvibe-host-deploy.lock';++const MIB = 1024 * 1024;++const { LockedError } = require('./lock');++++async function acquire(ctx, svc) {++ const { exec, inv, log } = ctx;++ const seconds = inv.deployGuard.lockWaitSeconds;++ try {++ return await exec.acquireDeployLock(LOCK_FILE, svc.id, seconds, (owner) => {++ log(`waiting for ${owner || 'another service'}'s deploy to finish (up to ${seconds}s)`);++ });++ } catch (err) { throw new LockedError(err.message); }++}++++/** Release the host lock only while the protected-session poll waits, then take it back. */++async function waitOutsideLock(ctx, svc, held, work) {++ await held.current.release();++ held.current = null;++ try { return await work(); }++ finally { held.current = await acquire(ctx, svc); }++}++++async function enoughMemory(exec, config, log = () => {}) {++ const { minAvailableMiB, memoryWaitSeconds, memoryPollSeconds } = config;++ const min = minAvailableMiB * MIB;++ const start = exec.now();++ let announced = false;++ for (;;) {++ const bytes = await exec.memAvailableBytes();++ if (bytes == null) throw new Error('cannot read MemAvailable from /proc/meminfo; refusing dependency install');++ if (bytes >= min) {++ if (announced) log(`MemAvailable is ${Math.floor(bytes / MIB)} MiB; proceeding with dependency install`);++ return;++ }++ if (!announced) {++ log(`MemAvailable is ${Math.floor(bytes / MIB)} MiB, below ${minAvailableMiB} MiB; waiting up to ${memoryWaitSeconds}s before dependency install`);++ announced = true;++ }++ if (exec.now() - start >= memoryWaitSeconds * 1000) throw new Error(`MemAvailable stayed below ${minAvailableMiB} MiB for ${memoryWaitSeconds}s; dependency install was not started`);++ await exec.sleep(Math.min(memoryPollSeconds * 1000, memoryWaitSeconds * 1000 - (exec.now() - start)));++ }++}++++module.exports = { LOCK_FILE, acquire, waitOutsideLock, enoughMemory };@@@ -10,6 +10,7 @@* resolve, the deploy aborts with nothing restarted.*/const path = require('path');++const deployGuard = require('./deploy-guard');/** Whether a package dir is excluded by skipPackages ("apps/_shared", or a prefix such as "apps/_*"). */function skipped(dir, patterns = []) {@@@ -59,7 +60,8 @@ async function needsInstall(g, from, to, pkgDir, lockfile = 'package-lock.json')* npm install as the checkout owner. Afterwards the lockfile is restored from git: npm rewrites it* on the host (different npm version), and a dirty lockfile blocks the next fast-forward pull.*/--async function install(exec, svc, g, pkgDir) {++async function install(exec, svc, g, pkgDir, log = () => {}) {++ await deployGuard.enoughMemory(exec, svc.deployGuard, log);const [cmd, ...args] = svc.install.command;const r = await exec.run(cmd, args, { as: svc.owner, cwd: path.join(svc.repo, pkgDir), timeoutMs: 20 * 60 * 1000 });if (r.code !== 0) throw new Error(`${svc.install.command.join(' ')} failed in ${pkgDir}: ${(r.stderr || r.stdout).trim().split('\n').slice(-3).join(' / ')}`);@@@ -103,7 +105,7 @@ async function verifyAndRepair(exec, svc, g, pkgDirs, log = () => {}) {}// A workspace installs once, at its root (pnpm).const at = svc.install.workspace ? '.' : pkgDir;-- if (!repaired.includes(at)) { await install(exec, svc, g, at); repaired.push(at); }++ if (!repaired.includes(at)) { await install(exec, svc, g, at, log); repaired.push(at); }}problems = await verify(exec, svc, pkgDirs);return { problems, repaired };@@@ -28,6 +28,8 @@* listeners(port) -> [{ pid, process }]* kill(pid, signal) -> bool (signal one pid; used only for a restore drill's own instance)* sleep(ms) ; now() -> ms ; isRoot() ; userName() ; hostname()++ * acquireDeployLock(file, service, timeoutSeconds, onWait) -> { release() } (kernel flock)++ * memAvailableBytes() -> number | null (Linux MemAvailable)*/const { spawn } = require('child_process');const fs = require('fs');@@@ -117,6 +119,44 @@ function createSystemExecutor({ env = process.env } = {}) {});}++ function acquireDeployLock(file, service, timeoutSeconds, onWait = () => {}) {++ return new Promise((resolve, reject) => {++ // The child owns the flock. Its stdin remains open for the whole deploy; if ovhost dies,++ // the pipe closes and the kernel releases the lock when the child exits.++ const child = spawn('flock', ['-x', '-w', String(timeoutSeconds), file, 'sh', '-c', 'printf "READY\\n"; cat >/dev/null'], { stdio: ['pipe', 'pipe', 'pipe'] });++ let ready = false;++ let closed = false;++ let stderr = '';++ const waiting = setTimeout(() => {++ if (!ready) {++ let owner = null;++ try { owner = fs.readFileSync(file, 'utf8').trim() || null; } catch { /* lock file may not exist yet */ }++ onWait(owner);++ }++ }, 250);++ child.stderr.on('data', (chunk) => { stderr += chunk; });++ child.stdout.on('data', () => {++ if (ready) return;++ ready = true;++ clearTimeout(waiting);++ try { fs.writeFileSync(file, `${service}\n`); }++ catch (err) { child.stdin.end(); reject(err); return; }++ resolve({ release: () => new Promise((done) => {++ // Leave the inode in place: unlinking a flock file can create two independent locks.++ if (closed) { done(); return; }++ child.once('close', done);++ child.stdin.end();++ }) });++ });++ child.on('error', (err) => { clearTimeout(waiting); if (!ready) reject(err); });++ child.on('close', (code) => {++ closed = true;++ clearTimeout(waiting);++ if (!ready) reject(new Error(`timed out waiting for the host deploy lock after ${timeoutSeconds}s${stderr.trim() ? `: ${stderr.trim()}` : ''} (flock exit ${code})`));++ });++ });++ }++async function readFile(p, { privileged } = {}) {try {return fs.readFileSync(p, 'utf8');@@@ -305,6 +345,11 @@ function createSystemExecutor({ env = process.env } = {}) {if (!Number.isInteger(pid) || pid <= 1) throw new Error(`refusing to signal pid ${pid}`);try { process.kill(pid, signal); return true; } catch (err) { if (err.code === 'ESRCH') return false; throw err; }},++ acquireDeployLock,++ memAvailableBytes: async () => {++ const m = /^MemAvailable:\s+(\d+)\s+kB$/m.exec(fs.readFileSync('/proc/meminfo', 'utf8'));++ return m ? Number(m[1]) * 1024 : null;++ },sleep: (ms) => new Promise((r) => setTimeout(r, ms)),now: () => Date.now(),isRoot: async () => isRoot(),@@@ -516,9 +516,16 @@ function normaliseNodes(raw) {function normalise(raw) {if (!raw || typeof raw !== 'object' || !raw.services || typeof raw.services !== 'object') throw new InventoryError('inventory needs a "services" object');const defaults = raw.defaults || {};++ const guard = { minAvailableMiB: 1536, memoryWaitSeconds: 300, memoryPollSeconds: 5, lockWaitSeconds: 1800, ...(raw.deployGuard || {}) };++ for (const [key, value] of Object.entries(guard)) {++ if (!['minAvailableMiB', 'memoryWaitSeconds', 'memoryPollSeconds', 'lockWaitSeconds'].includes(key) || !Number.isSafeInteger(value) || value <= 0) {++ throw new InventoryError(`deployGuard.${key} must be a positive integer`);++ }++ }const inv = {host: raw.host || null,stateDir: abs(raw.stateDir || '/var/lib/openvibe-host', 'stateDir'),++ deployGuard: guard,backupDir: abs(raw.backupDir || '/var/backups/openvibe', 'backupDir'),// Where postgres writes logical dumps before root takes the copies over (see// commands/backup.js). The same filesystem as backupDir, so the final move is a rename.@@@ -541,6 +548,7 @@ function normalise(raw) {for (const [id, svc] of Object.entries(raw.services)) {if (id.startsWith('_')) continue;inv.services[id] = normaliseService(id, svc, defaults);++ inv.services[id].deployGuard = guard;}const ports = new Map(Object.values(inv.services).filter((s) => s.port).map((s) => [s.port, s.id]));const drillPorts = new Map();@@@ -47,6 +47,7 @@ const probes = require('./probes');const readiness = require('./readiness');const releases = require('./releases');const lock = require('./lock');++const deployGuard = require('./deploy-guard');const preflight = require('./preflight');const ops = require('./release-ops');@@@ -382,7 +383,7 @@ async function prepare(ctx, svc, plan, record) {for (const i of plan.installs) {if (i.install) {log(`installing dependencies in the new release${i.pkg === '.' ? '' : ` (${i.pkg})`} (${[i.lockfileChanged && 'lockfile changed', i.depsChanged && 'package.json dependencies changed', i.missingNodeModules && `no node_modules in ${plan.fromRelease}`, !svc.release.reuseModules && 'always'].filter(Boolean).join(', ') || 'install'}) — nothing interrupted yet`);-- await deps.install(exec, rsvc, gitRelease(exec, svc, dir), i.pkg);++ await deps.install(exec, rsvc, gitRelease(exec, svc, dir), i.pkg, log);record.installed.push(i.pkg);} else {log(`dependencies unchanged — hard-linking node_modules from ${plan.fromRelease}${i.pkg === '.' ? '' : ` (${i.pkg})`}`);@@@ -416,6 +417,8 @@ async function apply(ctx, svc, mode, opts = {}) {if (!svc.managed) throw new OpError(`${svc.id} is not managed by ovhost${svc.unmanagedReason ? `: ${svc.unmanagedReason}` : ''}`, EXIT.USAGE);const held = await lock.acquire(exec, inv, svc.id);if (held.recoveredStale) log(`removed a stale lock (${held.file})`);++ const hostLock = { current: null };++ const guardedOpts = { ...opts, onIdleWait: (work) => deployGuard.waitOutsideLock(ctx, svc, hostLock, work) };const record = {id: releases.newId(exec.now()),service: svc.id,@@@ -442,6 +445,7 @@ async function apply(ctx, svc, mode, opts = {}) {};let writeRecord = true;try {++ hostLock.current = await deployGuard.acquire(ctx, svc);const plan = await computePlan(ctx, svc, { mode, to: opts.to, fetch: opts.fetch !== false, restart: opts.restart });Object.assign(record, { from: plan.from, to: plan.to, fromRelease: plan.fromRelease, toRelease: plan.toRelease, commits: plan.commits.length, lockfileChanged: plan.lockfileChanged, restartNeeded: plan.restartNeeded });log(`${svc.id}: ${mode} ${plan.fromRelease} (${plan.from.slice(0, 12)}) → ${mode === 'deploy' && !plan.reuse ? `new release ${plan.toRelease}` : plan.toRelease} (${plan.to.slice(0, 12)}; ${plan.commits.length} commit(s), ${plan.changedFiles} file(s) changed)`);@@@ -507,10 +511,10 @@ async function apply(ctx, svc, mode, opts = {}) {// A restart: sessions first, then the backup, then the switch. Refused or given up: the new// release is removed again and the current one never stopped serving.try {-- await ops.guardSessions(ctx, svc, opts, record, 'before the switch');++ await ops.guardSessions(ctx, svc, guardedOpts, record, 'before the switch');if (plan.backupNeeded) {await ops.backupBeforeRestart(ctx, svc, record);-- await ops.guardSessions(ctx, svc, opts, record, 'immediately before the restart');++ await ops.guardSessions(ctx, svc, guardedOpts, record, 'immediately before the restart');}} catch (err) {if (record.prepared) {@@@ -573,6 +577,7 @@ async function apply(ctx, svc, mode, opts = {}) {if (writeRecord) {try { await releases.append(exec, inv, record); } catch (e) { log(`warning: could not append to the release log: ${e.message}`); }}++ if (hostLock.current) await hostLock.current.release();await held.release();}}@@@ -43,6 +43,7 @@ const probes = require('./probes');const readiness = require('./readiness');const releases = require('./releases');const lock = require('./lock');++const deployGuard = require('./deploy-guard');const nginx = require('./nginx');const preflight = require('./preflight');@@@ -286,19 +287,22 @@ async function guardSessions(ctx, svc, opts, record, when) {let quiet = 0;let last = first;log(`${describe(first)}; --wait-idle: holding the restart (${when}); need ${svc.drain.quietChecks} idle checks ${svc.drain.pollSeconds}s apart, max ${svc.drain.waitMaxSeconds}s`);-- while (exec.now() - start < maxMs) {-- await exec.sleep(pollMs);-- last = await probes.countProtected(exec, svc);-- if (last.notRunning || last.count === 0) quiet += 1; else quiet = 0;-- if (quiet >= svc.drain.quietChecks) {-- record.protectedSessions.waitedSeconds = Math.round((exec.now() - start) / 1000);-- record.protectedSessions.count = 0;-- log(`idle for ${quiet} checks — proceeding (waited ${record.protectedSessions.waitedSeconds}s)`);-- return;++ const poll = async () => {++ while (exec.now() - start < maxMs) {++ await exec.sleep(pollMs);++ last = await probes.countProtected(exec, svc);++ if (last.notRunning || last.count === 0) quiet += 1; else quiet = 0;++ if (quiet >= svc.drain.quietChecks) {++ record.protectedSessions.waitedSeconds = Math.round((exec.now() - start) / 1000);++ record.protectedSessions.count = 0;++ log(`idle for ${quiet} checks — proceeding (waited ${record.protectedSessions.waitedSeconds}s)`);++ return;++ }++ log(`${describe(last)}; still holding (${Math.round((exec.now() - start) / 1000)}s)`);}-- log(`${describe(last)}; still holding (${Math.round((exec.now() - start) / 1000)}s)`);-- }-- throw new OpError(`${describe(last)} after waiting ${svc.drain.waitMaxSeconds}s — gave up; nothing was restarted`, EXIT.PROTECTED, { result: 'gave-up-waiting' });++ throw new OpError(`${describe(last)} after waiting ${svc.drain.waitMaxSeconds}s — gave up; nothing was restarted`, EXIT.PROTECTED, { result: 'gave-up-waiting' });++ };++ return opts.onIdleWait ? opts.onIdleWait(poll) : poll();}// ── helpers ──────────────────────────────────────────────────────────────────@@@ -307,7 +311,7 @@ async function installAll(ctx, svc, g, installs, record) {for (const i of installs) {if (!i.install) continue;ctx.log(`installing dependencies in ${i.pkg} (${[i.lockfileChanged && 'lockfile changed', i.depsChanged && 'package.json dependencies changed', i.missingNodeModules && 'no node_modules', svc.install.always && 'always'].filter(Boolean).join(', ')})`);-- await deps.install(ctx.exec, svc, g, i.pkg);++ await deps.install(ctx.exec, svc, g, i.pkg, ctx.log);record.installed.push(i.pkg);}}@@@ -395,7 +399,7 @@ async function restoreCheckout(ctx, svc, g, sha, record) {if (record.installed.length) {for (const pkg of record.installed) {ctx.log(`reinstalling dependencies in ${pkg} for ${sha.slice(0, 12)}`);-- await deps.install(ctx.exec, svc, g, pkg);++ await deps.install(ctx.exec, svc, g, pkg, ctx.log);}}const v = await verifyDeps(ctx, svc, g);@@@ -440,6 +444,8 @@ async function apply(ctx, svc, mode, opts = {}) {}const held = await lock.acquire(exec, inv, svc.id);if (held.recoveredStale) log(`removed a stale lock (${held.file})`);++ const hostLock = { current: null };++ const guardedOpts = { ...opts, onIdleWait: (work) => deployGuard.waitOutsideLock(ctx, svc, hostLock, work) };const g = git(exec, svc);const record = {id: releases.newId(exec.now()),@@@ -464,6 +470,7 @@ async function apply(ctx, svc, mode, opts = {}) {let exitCode = EXIT.OK;let writeRecord = true;try {++ hostLock.current = await deployGuard.acquire(ctx, svc);const plan = await computePlan(ctx, svc, { mode, to: opts.to, fetch: opts.fetch !== false, restart: opts.restart });Object.assign(record, { from: plan.from, to: plan.to, commits: plan.commits.length, lockfileChanged: plan.lockfileChanged, restartNeeded: plan.restartNeeded });log(`${svc.id}: ${mode} ${plan.from.slice(0, 12)} → ${plan.to.slice(0, 12)} (${plan.commits.length} commit(s), ${plan.changedFiles} file(s) changed)`);@@@ -487,7 +494,7 @@ async function apply(ctx, svc, mode, opts = {}) {if (plan.unitDrift.length && !opts.installUnits && !svc.installUnits) log(`note: unit files differ from the repo (${plan.unitDrift.map((u) => u.unit).join(', ')}); pass --install-units to install them`);// Sessions first: in place, the running process would otherwise see its files change under it.-- if (plan.restartNeeded) await guardSessions(ctx, svc, opts, record, 'before the checkout moves');++ if (plan.restartNeeded) await guardSessions(ctx, svc, guardedOpts, record, 'before the checkout moves');if (plan.generatedDirty.length && !plan.upToDate) {log(`restoring ${plan.generatedDirty.length} tracked build output file(s) from git (${svc.generated.join(', ')})`);@@@ -538,7 +545,7 @@ async function apply(ctx, svc, mode, opts = {}) {}if (opts.installUnits || svc.installUnits) await installUnits(ctx, svc, record);if (plan.backupNeeded && plan.restartNeeded) await backupBeforeRestart(ctx, svc, record);-- if (plan.restartNeeded) await guardSessions(ctx, svc, opts, record, 'immediately before the restart');++ if (plan.restartNeeded) await guardSessions(ctx, svc, guardedOpts, record, 'immediately before the restart');} catch (err) {// Nothing has been restarted: put the files back the way the running process expects.if (record.checkoutMoved) {@@@ -606,6 +613,7 @@ async function apply(ctx, svc, mode, opts = {}) {if (writeRecord) {try { await releases.append(exec, inv, record); } catch (e) { log(`warning: could not append to the release log: ${e.message}`); }}++ if (hostLock.current) await hostLock.current.release();await held.release();}}nnew file mode 100644@@@ -0,0 +1,108 @@++'use strict';++const assert = require('assert');++const { scenario, push, test, runTests } = require('./helpers');++const { liveHost } = require('./strategy-hosts');++++const tick = () => new Promise((resolve) => setImmediate(resolve));++++runTests([++ test('concurrent deploys serialize preparation across services', async () => {++ const host = scenario();++ push(host, 'live', { 'package-lock.json': '{"lockfileVersion":3,"v":2}', 'server/index.js': 'v2();' });++ const oldMedia = host.repo('media').head;++ push(host, 'media', { 'server/index.js': 'v2();' });++ const originalRun = host.exec.run.bind(host.exec);++ let entered;++ const installing = new Promise((resolve) => { entered = resolve; });++ let continueInstall;++ const barrier = new Promise((resolve) => { continueInstall = resolve; });++ host.exec.run = async (cmd, args, opts) => {++ if (cmd === 'npm' && !host.installPaused) {++ host.installPaused = true;++ entered();++ await barrier;++ }++ return originalRun(cmd, args, opts);++ };++ const first = host.cli('deploy', 'live');++ await installing;++ const second = host.cli('deploy', 'media');++ await tick();++ assert.deepStrictEqual(host.deployLockEvents.slice(0, 2), ['acquire:live', 'wait:media:live']);++ assert.strictEqual(host.repo('media').head, oldMedia, 'media did not prepare while live installed');++ continueInstall();++ const [a, b] = await Promise.all([first, second]);++ assert.strictEqual(a.code, 0, a.out);++ assert.strictEqual(b.code, 0, b.out);++ assert.match(b.out, /waiting for live's deploy to finish/);++ assert.deepStrictEqual(host.deployLockEvents, ['acquire:live', 'wait:media:live', 'release:live', 'acquire:media', 'release:media']);++ }),++++ test('--wait-idle releases the host lock while streams are active', async () => {++ const host = scenario();++ push(host, 'live', { 'server/index.js': 'v2();' });++ push(host, 'media', { 'server/index.js': 'v2();' });++ let probes = 0;++ host.http.set('http://127.0.0.1:3000/api/streams', () => ({ status: 200, body: { streams: ++probes <= 2 ? [{ is_live: true }] : [] } }));++ const originalSleep = host.exec.sleep.bind(host.exec);++ let sleeping;++ const entered = new Promise((resolve) => { sleeping = resolve; });++ let resume;++ const barrier = new Promise((resolve) => { resume = resolve; });++ host.exec.sleep = async (ms) => {++ if (ms === 60000 && !host.waitPaused) { host.waitPaused = true; sleeping(); await barrier; }++ return originalSleep(ms);++ };++ const first = host.cli('deploy', 'live', '--wait-idle');++ await entered;++ assert.deepStrictEqual(host.deployLockEvents.slice(0, 2), ['acquire:live', 'release:live']);++ const other = await host.cli('deploy', 'media');++ assert.strictEqual(other.code, 0, other.out);++ resume();++ const result = await first;++ assert.strictEqual(result.code, 0, result.out);++ assert.deepStrictEqual(host.deployLockEvents, ['acquire:live', 'release:live', 'acquire:media', 'release:media', 'acquire:live', 'release:live']);++ }),++++ test('memory guard waits for MemAvailable and then installs', async () => {++ const host = scenario();++ push(host, 'live', { 'package-lock.json': '{"lockfileVersion":3,"v":2}' });++ const readings = [512, 768, 2048];++ host.memAvailableMiB = () => readings.shift() || 2048;++ const r = await host.cli('deploy', 'live');++ assert.strictEqual(r.code, 0, r.out);++ assert.match(r.out, /MemAvailable is 512 MiB, below 1536 MiB/);++ assert.match(r.out, /MemAvailable is 2048 MiB; proceeding/);++ assert.ok(host.calls.some((c) => c.cmd === 'npm'));++ }),++++ test('memory guard times out without starting npm or restarting', async () => {++ const host = scenario();++ const from = host.repo('live').head;++ push(host, 'live', { 'package-lock.json': '{"lockfileVersion":3,"v":2}' });++ host.memAvailableMiB = 512;++ const inventory = JSON.parse(host.read('/etc/openvibe/host.json'));++ inventory.deployGuard = { memoryWaitSeconds: 10 };++ host.put('/etc/openvibe/host.json', JSON.stringify(inventory), { mode: 0o640, owner: 'root' });++ const r = await host.cli('deploy', 'live');++ assert.strictEqual(r.code, 2, r.out);++ assert.match(r.out, /MemAvailable stayed below 1536 MiB for 10s/);++ assert.strictEqual(host.repo('live').head, from);++ assert.ok(!host.calls.some((c) => c.cmd === 'npm'));++ assert.deepStrictEqual(host.restarts(), []);++ }),++ test('release-layout memory timeout removes the prepared release', async () => {++ const host = await liveHost();++ host.push({ 'package-lock.json': '{"lockfileVersion":3,"v":2}' }, 'dependencies');++ host.memAvailableMiB = 512;++ const inventory = JSON.parse(host.read('/etc/openvibe/host.json'));++ inventory.deployGuard = { memoryWaitSeconds: 10 };++ host.put('/etc/openvibe/host.json', JSON.stringify(inventory), { mode: 0o640, owner: 'root' });++ const result = await host.cli('deploy', 'live');++ assert.strictEqual(result.code, 2, result.out);++ assert.deepStrictEqual(await host.releaseIds(), [host.firstRelease]);++ assert.strictEqual(host.current(), host.firstRelease);++ assert.ok(!host.calls.some((c) => c.cmd === 'npm'));++ assert.deepStrictEqual(host.restarts(), []);++ }),++]);@@@ -15,6 +15,20 @@ const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'ovhost-exec-'));process.on('exit', () => fs.rmSync(dir, { recursive: true, force: true }));runTests([++ test('kernel flock serializes two holders and reports the service that owns it', async () => {++ const file = path.join(dir, 'deploy.lock');++ const first = await exec.acquireDeployLock(file, 'live', 2);++ let waiting = null;++ const secondPromise = exec.acquireDeployLock(file, 'media', 2, (owner) => { waiting = owner; });++ await new Promise((resolve) => setTimeout(resolve, 350));++ assert.strictEqual(waiting, 'live');++ await assert.rejects(exec.acquireDeployLock(file, 'third', 0.1), /timed out waiting for the host deploy lock/);++ await first.release();++ const second = await secondPromise;++ assert.strictEqual(fs.readFileSync(file, 'utf8').trim(), 'media');++ await second.release();++ assert.ok((await exec.memAvailableBytes()) > 0);++ }),test('http, files, locks and run() behave as the fake host assumes', async () => {const server = http.createServer((req, res) => { res.writeHead(req.url === '/ok' ? 200 : 503); res.end(JSON.stringify({ host: req.headers.host })); });await new Promise((r) => server.listen(0, '127.0.0.1', r));@@@ -24,10 +24,14 @@ function createFakeHost({ root = true, user = 'root', hostname = 'fake-host', stconst http = new Map(); // url -> fn({ headers, method, body }) -> { status, body } (request() also passes method and body)const alivePids = new Set();const listeners = new Map(); // port -> [{ pid, process }]++ const deployLockQueue = [];++ let deployLockOwner = null;let clock = start;let seq = 0;const host = {files, calls, repos, remoteRepos, units, http, alivePids, listeners,++ deployLockEvents: [],++ memAvailableMiB: 4096,nginxTest: () => ({ code: 0, stderr: 'nginx: configuration file /etc/nginx/nginx.conf test is successful' }),npm: null, // (cwd, argv, as) -> undefined | { code, stderr } ; default installs package.json depsnpmRewritesLockfile: true,@@@ -765,6 +769,21 @@ function createFakeHost({ root = true, user = 'root', hostname = 'fake-host', st}return true;},++ async acquireDeployLock(file, service, timeoutSeconds, onWait) {++ if (deployLockOwner) {++ host.deployLockEvents.push(`wait:${service}:${deployLockOwner}`);++ onWait(deployLockOwner);++ await new Promise((resolve) => deployLockQueue.push(resolve));++ }++ deployLockOwner = service;++ host.deployLockEvents.push(`acquire:${service}`);++ return { release: async () => {++ host.deployLockEvents.push(`release:${service}`);++ deployLockOwner = null;++ if (deployLockQueue.length) deployLockQueue.shift()();++ } };++ },++ async memAvailableBytes() { return (typeof host.memAvailableMiB === 'function' ? host.memAvailableMiB() : host.memAvailableMiB) * 1024 * 1024; },async sleep(ms) { clock += ms; },now: () => clock,async isRoot() { return root; },
@@ -19,6 +19,12 @@ Production deploys and rollbacks use `ovhost deploy <svc>` and `ovhost rollback
@@ -19,6 +19,12 @@ Production deploys and rollbacks use `ovhost deploy <svc>` and `ovhost rollback
Every other service keeps the default strategy, `git-checkout`, which is what `ovhost deploy` did before.
Every other service keeps the default strategy, `git-checkout`, which is what `ovhost deploy` did before.
+## Host deploy admission
+
+Deploys and rollbacks take one host-wide kernel lock at `/run/lock/openvibe-host-deploy.lock` before preparing a release or changing a checkout. The lock remains held through preflight, switching, restart, readiness and automatic rollback. Another service waits and names the current owner; it fails after the configured lock timeout. A `--wait-idle` operation releases the host lock while it polls streams or recordings, then takes it again before continuing. The existing per-service lock remains held during that wait.
+
+Before each dependency install, ovhost checks Linux `MemAvailable`. When it is below 1536 MiB, ovhost waits for memory and reports the wait. If memory stays low for 300 seconds, the install does not start and the deploy fails through its normal cleanup path. The root of `host.json` may set `deployGuard.minAvailableMiB`, `deployGuard.memoryWaitSeconds`, `deployGuard.memoryPollSeconds`, and `deployGuard.lockWaitSeconds` (default 1536, 300, 5, and 1800). Each must be a positive integer. This applies to `npm` and `pnpm` installs in both deploy strategies and to dependency reinstall during automatic rollback.
+
## Deploy wrappers
## Deploy wrappers
Repository wrappers call `ovhost plan`, `ovhost deploy`, or `ovhost rollback` for their service. If ovhost is missing, the wrapper exits with an error. The `capabilities` command remains available for installed wrappers that still query it.
Repository wrappers call `ovhost plan`, `ovhost deploy`, or `ovhost rollback` for their service. If ovhost is missing, the wrapper exits with an error. The `capabilities` command remains available for installed wrappers that still query it.
@@ -364,6 +364,18 @@ const SCENARIOS = [
@@ -364,6 +364,18 @@ const SCENARIOS = [
budget: { name: 'calls dropped by a deploy', max: 0 },
budget: { name: 'calls dropped by a deploy', max: 0 },
reason: 'no proof: Chat\'s drain policy is report (a deploy restarts with calls up); a restart ends them with end_reason restart (6e) and people call again',
reason: 'no proof: Chat\'s drain policy is report (a deploy restarts with calls up); a restart ends them with end_reason restart (6e) and people call again',
},
},
+ {
+ id: '6j', what: 'host-wide deploy locking serializes preparation, while --wait-idle frees the lock', repo: 'host', files: ['test/deploy-guard.test.js'],
+ expect: [ok('concurrent deploys serialize preparation across services'), ok('--wait-idle releases the host lock while streams are active')],
+ budget: { name: 'overlapping deploy preparations', max: 0 }, measure: asserted(), category: 'resource growth',
+ source: 'fake host asserts lock acquisition order and that the second checkout does not move during the first install',
+ },
+ {
+ id: '6k', what: 'low MemAvailable delays dependency install and times out cleanly', repo: 'host', files: ['test/deploy-guard.test.js'],
+ expect: [ok('memory guard waits for MemAvailable and then installs'), ok('memory guard times out without starting npm or restarting'), ok('release-layout memory timeout removes the prepared release')],
+ budget: { name: 'installs begun below the memory floor', max: 0 }, measure: asserted(), category: 'resource growth',
+ source: 'fake host varies MemAvailable and asserts npm and restart are not called on timeout',
+ },
],
],
},
},
{
{
new file mode 100644
new file mode 100644
@@ -0,0 +1,47 @@
@@ -0,0 +1,47 @@
+'use strict';
+/** Host-wide deploy admission. The executor holds a kernel flock until release(). */
+
+const LOCK_FILE = '/run/lock/openvibe-host-deploy.lock';
+const MIB = 1024 * 1024;
+const { LockedError } = require('./lock');
+
+async function acquire(ctx, svc) {
+ const { exec, inv, log } = ctx;
+ const seconds = inv.deployGuard.lockWaitSeconds;
+ try {
+ return await exec.acquireDeployLock(LOCK_FILE, svc.id, seconds, (owner) => {
+ log(`waiting for ${owner || 'another service'}'s deploy to finish (up to ${seconds}s)`);
+ });
+ } catch (err) { throw new LockedError(err.message); }
+}
+
+/** Release the host lock only while the protected-session poll waits, then take it back. */
+async function waitOutsideLock(ctx, svc, held, work) {
+ await held.current.release();
+ held.current = null;
+ try { return await work(); }
+ finally { held.current = await acquire(ctx, svc); }
+}
+
+async function enoughMemory(exec, config, log = () => {}) {
+ const { minAvailableMiB, memoryWaitSeconds, memoryPollSeconds } = config;
+ const min = minAvailableMiB * MIB;
+ const start = exec.now();
+ let announced = false;
+ for (;;) {
+ const bytes = await exec.memAvailableBytes();
+ if (bytes == null) throw new Error('cannot read MemAvailable from /proc/meminfo; refusing dependency install');
+ if (bytes >= min) {
+ if (announced) log(`MemAvailable is ${Math.floor(bytes / MIB)} MiB; proceeding with dependency install`);
+ return;
+ }
+ if (!announced) {
+ log(`MemAvailable is ${Math.floor(bytes / MIB)} MiB, below ${minAvailableMiB} MiB; waiting up to ${memoryWaitSeconds}s before dependency install`);
+ announced = true;
+ }
+ if (exec.now() - start >= memoryWaitSeconds * 1000) throw new Error(`MemAvailable stayed below ${minAvailableMiB} MiB for ${memoryWaitSeconds}s; dependency install was not started`);
+ await exec.sleep(Math.min(memoryPollSeconds * 1000, memoryWaitSeconds * 1000 - (exec.now() - start)));
+ }
+}
+
+module.exports = { LOCK_FILE, acquire, waitOutsideLock, enoughMemory };
@@ -10,6 +10,7 @@
@@ -10,6 +10,7 @@
* resolve, the deploy aborts with nothing restarted.
* resolve, the deploy aborts with nothing restarted.
*/
*/
const path = require('path');
const path = require('path');
+const deployGuard = require('./deploy-guard');
/** Whether a package dir is excluded by skipPackages ("apps/_shared", or a prefix such as "apps/_*"). */
/** Whether a package dir is excluded by skipPackages ("apps/_shared", or a prefix such as "apps/_*"). */
function skipped(dir, patterns = []) {
function skipped(dir, patterns = []) {
@@ -59,7 +60,8 @@ async function needsInstall(g, from, to, pkgDir, lockfile = 'package-lock.json')
@@ -59,7 +60,8 @@ async function needsInstall(g, from, to, pkgDir, lockfile = 'package-lock.json')
* npm install as the checkout owner. Afterwards the lockfile is restored from git: npm rewrites it
* npm install as the checkout owner. Afterwards the lockfile is restored from git: npm rewrites it
* on the host (different npm version), and a dirty lockfile blocks the next fast-forward pull.
* on the host (different npm version), and a dirty lockfile blocks the next fast-forward pull.
*/
*/
-async function install(exec, svc, g, pkgDir) {
+async function install(exec, svc, g, pkgDir, log = () => {}) {
+ await deployGuard.enoughMemory(exec, svc.deployGuard, log);
const [cmd, ...args] = svc.install.command;
const [cmd, ...args] = svc.install.command;
const r = await exec.run(cmd, args, { as: svc.owner, cwd: path.join(svc.repo, pkgDir), timeoutMs: 20 * 60 * 1000 });
const r = await exec.run(cmd, args, { as: svc.owner, cwd: path.join(svc.repo, pkgDir), timeoutMs: 20 * 60 * 1000 });
if (r.code !== 0) throw new Error(`${svc.install.command.join(' ')} failed in ${pkgDir}: ${(r.stderr || r.stdout).trim().split('\n').slice(-3).join(' / ')}`);
if (r.code !== 0) throw new Error(`${svc.install.command.join(' ')} failed in ${pkgDir}: ${(r.stderr || r.stdout).trim().split('\n').slice(-3).join(' / ')}`);
@@ -103,7 +105,7 @@ async function verifyAndRepair(exec, svc, g, pkgDirs, log = () => {}) {
@@ -103,7 +105,7 @@ async function verifyAndRepair(exec, svc, g, pkgDirs, log = () => {}) {
}
}
// A workspace installs once, at its root (pnpm).
// A workspace installs once, at its root (pnpm).
const at = svc.install.workspace ? '.' : pkgDir;
const at = svc.install.workspace ? '.' : pkgDir;
- if (!repaired.includes(at)) { await install(exec, svc, g, at); repaired.push(at); }
+ if (!repaired.includes(at)) { await install(exec, svc, g, at, log); repaired.push(at); }
}
}
problems = await verify(exec, svc, pkgDirs);
problems = await verify(exec, svc, pkgDirs);
return { problems, repaired };
return { problems, repaired };
@@ -28,6 +28,8 @@
@@ -28,6 +28,8 @@
* listeners(port) -> [{ pid, process }]
* listeners(port) -> [{ pid, process }]
* kill(pid, signal) -> bool (signal one pid; used only for a restore drill's own instance)
* kill(pid, signal) -> bool (signal one pid; used only for a restore drill's own instance)
* sleep(ms) ; now() -> ms ; isRoot() ; userName() ; hostname()
* sleep(ms) ; now() -> ms ; isRoot() ; userName() ; hostname()
+ * acquireDeployLock(file, service, timeoutSeconds, onWait) -> { release() } (kernel flock)
+ * memAvailableBytes() -> number | null (Linux MemAvailable)
*/
*/
const { spawn } = require('child_process');
const { spawn } = require('child_process');
const fs = require('fs');
const fs = require('fs');
@@ -117,6 +119,44 @@ function createSystemExecutor({ env = process.env } = {}) {
@@ -117,6 +119,44 @@ function createSystemExecutor({ env = process.env } = {}) {
});
});
}
}
+ function acquireDeployLock(file, service, timeoutSeconds, onWait = () => {}) {
+ return new Promise((resolve, reject) => {
+ // The child owns the flock. Its stdin remains open for the whole deploy; if ovhost dies,
+ // the pipe closes and the kernel releases the lock when the child exits.
+ const child = spawn('flock', ['-x', '-w', String(timeoutSeconds), file, 'sh', '-c', 'printf "READY\\n"; cat >/dev/null'], { stdio: ['pipe', 'pipe', 'pipe'] });
+ let ready = false;
+ let closed = false;
+ let stderr = '';
+ const waiting = setTimeout(() => {
+ if (!ready) {
+ let owner = null;
+ try { owner = fs.readFileSync(file, 'utf8').trim() || null; } catch { /* lock file may not exist yet */ }
+ onWait(owner);
+ }
+ }, 250);
+ child.stderr.on('data', (chunk) => { stderr += chunk; });
+ child.stdout.on('data', () => {
+ if (ready) return;
+ ready = true;
+ clearTimeout(waiting);
+ try { fs.writeFileSync(file, `${service}\n`); }
+ catch (err) { child.stdin.end(); reject(err); return; }
+ resolve({ release: () => new Promise((done) => {
+ // Leave the inode in place: unlinking a flock file can create two independent locks.
+ if (closed) { done(); return; }
+ child.once('close', done);
+ child.stdin.end();
+ }) });
+ });
+ child.on('error', (err) => { clearTimeout(waiting); if (!ready) reject(err); });
+ child.on('close', (code) => {
+ closed = true;
+ clearTimeout(waiting);
+ if (!ready) reject(new Error(`timed out waiting for the host deploy lock after ${timeoutSeconds}s${stderr.trim() ? `: ${stderr.trim()}` : ''} (flock exit ${code})`));
+ });
+ });
+ }
+
async function readFile(p, { privileged } = {}) {
async function readFile(p, { privileged } = {}) {
try {
try {
return fs.readFileSync(p, 'utf8');
return fs.readFileSync(p, 'utf8');
@@ -305,6 +345,11 @@ function createSystemExecutor({ env = process.env } = {}) {
@@ -305,6 +345,11 @@ function createSystemExecutor({ env = process.env } = {}) {
if (!Number.isInteger(pid) || pid <= 1) throw new Error(`refusing to signal pid ${pid}`);
if (!Number.isInteger(pid) || pid <= 1) throw new Error(`refusing to signal pid ${pid}`);
try { process.kill(pid, signal); return true; } catch (err) { if (err.code === 'ESRCH') return false; throw err; }
try { process.kill(pid, signal); return true; } catch (err) { if (err.code === 'ESRCH') return false; throw err; }
},
},
+ acquireDeployLock,
+ memAvailableBytes: async () => {
+ const m = /^MemAvailable:\s+(\d+)\s+kB$/m.exec(fs.readFileSync('/proc/meminfo', 'utf8'));
+ return m ? Number(m[1]) * 1024 : null;
+ },
sleep: (ms) => new Promise((r) => setTimeout(r, ms)),
sleep: (ms) => new Promise((r) => setTimeout(r, ms)),
now: () => Date.now(),
now: () => Date.now(),
isRoot: async () => isRoot(),
isRoot: async () => isRoot(),
@@ -516,9 +516,16 @@ function normaliseNodes(raw) {
@@ -516,9 +516,16 @@ function normaliseNodes(raw) {
function normalise(raw) {
function normalise(raw) {
if (!raw || typeof raw !== 'object' || !raw.services || typeof raw.services !== 'object') throw new InventoryError('inventory needs a "services" object');
if (!raw || typeof raw !== 'object' || !raw.services || typeof raw.services !== 'object') throw new InventoryError('inventory needs a "services" object');
const defaults = raw.defaults || {};
const defaults = raw.defaults || {};
+ const guard = { minAvailableMiB: 1536, memoryWaitSeconds: 300, memoryPollSeconds: 5, lockWaitSeconds: 1800, ...(raw.deployGuard || {}) };
+ for (const [key, value] of Object.entries(guard)) {
+ if (!['minAvailableMiB', 'memoryWaitSeconds', 'memoryPollSeconds', 'lockWaitSeconds'].includes(key) || !Number.isSafeInteger(value) || value <= 0) {
+ throw new InventoryError(`deployGuard.${key} must be a positive integer`);
+ }
+ }
const inv = {
const inv = {
host: raw.host || null,
host: raw.host || null,
stateDir: abs(raw.stateDir || '/var/lib/openvibe-host', 'stateDir'),
stateDir: abs(raw.stateDir || '/var/lib/openvibe-host', 'stateDir'),
+ deployGuard: guard,
backupDir: abs(raw.backupDir || '/var/backups/openvibe', 'backupDir'),
backupDir: abs(raw.backupDir || '/var/backups/openvibe', 'backupDir'),
// Where postgres writes logical dumps before root takes the copies over (see
// Where postgres writes logical dumps before root takes the copies over (see
// commands/backup.js). The same filesystem as backupDir, so the final move is a rename.
// commands/backup.js). The same filesystem as backupDir, so the final move is a rename.
@@ -541,6 +548,7 @@ function normalise(raw) {
@@ -541,6 +548,7 @@ function normalise(raw) {
for (const [id, svc] of Object.entries(raw.services)) {
for (const [id, svc] of Object.entries(raw.services)) {
if (id.startsWith('_')) continue;
if (id.startsWith('_')) continue;
inv.services[id] = normaliseService(id, svc, defaults);
inv.services[id] = normaliseService(id, svc, defaults);
+ inv.services[id].deployGuard = guard;
}
}
const ports = new Map(Object.values(inv.services).filter((s) => s.port).map((s) => [s.port, s.id]));
const ports = new Map(Object.values(inv.services).filter((s) => s.port).map((s) => [s.port, s.id]));
const drillPorts = new Map();
const drillPorts = new Map();
@@ -47,6 +47,7 @@ const probes = require('./probes');
@@ -47,6 +47,7 @@ const probes = require('./probes');
const readiness = require('./readiness');
const readiness = require('./readiness');
const releases = require('./releases');
const releases = require('./releases');
const lock = require('./lock');
const lock = require('./lock');
+const deployGuard = require('./deploy-guard');
const preflight = require('./preflight');
const preflight = require('./preflight');
const ops = require('./release-ops');
const ops = require('./release-ops');
@@ -382,7 +383,7 @@ async function prepare(ctx, svc, plan, record) {
@@ -382,7 +383,7 @@ async function prepare(ctx, svc, plan, record) {
for (const i of plan.installs) {
for (const i of plan.installs) {
if (i.install) {
if (i.install) {
log(`installing dependencies in the new release${i.pkg === '.' ? '' : ` (${i.pkg})`} (${[i.lockfileChanged && 'lockfile changed', i.depsChanged && 'package.json dependencies changed', i.missingNodeModules && `no node_modules in ${plan.fromRelease}`, !svc.release.reuseModules && 'always'].filter(Boolean).join(', ') || 'install'}) — nothing interrupted yet`);
log(`installing dependencies in the new release${i.pkg === '.' ? '' : ` (${i.pkg})`} (${[i.lockfileChanged && 'lockfile changed', i.depsChanged && 'package.json dependencies changed', i.missingNodeModules && `no node_modules in ${plan.fromRelease}`, !svc.release.reuseModules && 'always'].filter(Boolean).join(', ') || 'install'}) — nothing interrupted yet`);
- await deps.install(exec, rsvc, gitRelease(exec, svc, dir), i.pkg);
+ await deps.install(exec, rsvc, gitRelease(exec, svc, dir), i.pkg, log);
record.installed.push(i.pkg);
record.installed.push(i.pkg);
} else {
} else {
log(`dependencies unchanged — hard-linking node_modules from ${plan.fromRelease}${i.pkg === '.' ? '' : ` (${i.pkg})`}`);
log(`dependencies unchanged — hard-linking node_modules from ${plan.fromRelease}${i.pkg === '.' ? '' : ` (${i.pkg})`}`);
@@ -416,6 +417,8 @@ async function apply(ctx, svc, mode, opts = {}) {
@@ -416,6 +417,8 @@ async function apply(ctx, svc, mode, opts = {}) {
if (!svc.managed) throw new OpError(`${svc.id} is not managed by ovhost${svc.unmanagedReason ? `: ${svc.unmanagedReason}` : ''}`, EXIT.USAGE);
if (!svc.managed) throw new OpError(`${svc.id} is not managed by ovhost${svc.unmanagedReason ? `: ${svc.unmanagedReason}` : ''}`, EXIT.USAGE);
const held = await lock.acquire(exec, inv, svc.id);
const held = await lock.acquire(exec, inv, svc.id);
if (held.recoveredStale) log(`removed a stale lock (${held.file})`);
if (held.recoveredStale) log(`removed a stale lock (${held.file})`);
+ const hostLock = { current: null };
+ const guardedOpts = { ...opts, onIdleWait: (work) => deployGuard.waitOutsideLock(ctx, svc, hostLock, work) };
const record = {
const record = {
id: releases.newId(exec.now()),
id: releases.newId(exec.now()),
service: svc.id,
service: svc.id,
@@ -442,6 +445,7 @@ async function apply(ctx, svc, mode, opts = {}) {
@@ -442,6 +445,7 @@ async function apply(ctx, svc, mode, opts = {}) {
};
};
let writeRecord = true;
let writeRecord = true;
try {
try {
+ hostLock.current = await deployGuard.acquire(ctx, svc);
const plan = await computePlan(ctx, svc, { mode, to: opts.to, fetch: opts.fetch !== false, restart: opts.restart });
const plan = await computePlan(ctx, svc, { mode, to: opts.to, fetch: opts.fetch !== false, restart: opts.restart });
Object.assign(record, { from: plan.from, to: plan.to, fromRelease: plan.fromRelease, toRelease: plan.toRelease, commits: plan.commits.length, lockfileChanged: plan.lockfileChanged, restartNeeded: plan.restartNeeded });
Object.assign(record, { from: plan.from, to: plan.to, fromRelease: plan.fromRelease, toRelease: plan.toRelease, commits: plan.commits.length, lockfileChanged: plan.lockfileChanged, restartNeeded: plan.restartNeeded });
log(`${svc.id}: ${mode} ${plan.fromRelease} (${plan.from.slice(0, 12)}) → ${mode === 'deploy' && !plan.reuse ? `new release ${plan.toRelease}` : plan.toRelease} (${plan.to.slice(0, 12)}; ${plan.commits.length} commit(s), ${plan.changedFiles} file(s) changed)`);
log(`${svc.id}: ${mode} ${plan.fromRelease} (${plan.from.slice(0, 12)}) → ${mode === 'deploy' && !plan.reuse ? `new release ${plan.toRelease}` : plan.toRelease} (${plan.to.slice(0, 12)}; ${plan.commits.length} commit(s), ${plan.changedFiles} file(s) changed)`);
@@ -507,10 +511,10 @@ async function apply(ctx, svc, mode, opts = {}) {
@@ -507,10 +511,10 @@ async function apply(ctx, svc, mode, opts = {}) {
// A restart: sessions first, then the backup, then the switch. Refused or given up: the new
// A restart: sessions first, then the backup, then the switch. Refused or given up: the new
// release is removed again and the current one never stopped serving.
// release is removed again and the current one never stopped serving.
try {
try {
- await ops.guardSessions(ctx, svc, opts, record, 'before the switch');
+ await ops.guardSessions(ctx, svc, guardedOpts, record, 'before the switch');
if (plan.backupNeeded) {
if (plan.backupNeeded) {
await ops.backupBeforeRestart(ctx, svc, record);
await ops.backupBeforeRestart(ctx, svc, record);
- await ops.guardSessions(ctx, svc, opts, record, 'immediately before the restart');
+ await ops.guardSessions(ctx, svc, guardedOpts, record, 'immediately before the restart');
}
}
} catch (err) {
} catch (err) {
if (record.prepared) {
if (record.prepared) {
@@ -573,6 +577,7 @@ async function apply(ctx, svc, mode, opts = {}) {
@@ -573,6 +577,7 @@ async function apply(ctx, svc, mode, opts = {}) {
if (writeRecord) {
if (writeRecord) {
try { await releases.append(exec, inv, record); } catch (e) { log(`warning: could not append to the release log: ${e.message}`); }
try { await releases.append(exec, inv, record); } catch (e) { log(`warning: could not append to the release log: ${e.message}`); }
}
}
+ if (hostLock.current) await hostLock.current.release();
await held.release();
await held.release();
}
}
}
}
@@ -43,6 +43,7 @@ const probes = require('./probes');
@@ -43,6 +43,7 @@ const probes = require('./probes');
const readiness = require('./readiness');
const readiness = require('./readiness');
const releases = require('./releases');
const releases = require('./releases');
const lock = require('./lock');
const lock = require('./lock');
+const deployGuard = require('./deploy-guard');
const nginx = require('./nginx');
const nginx = require('./nginx');
const preflight = require('./preflight');
const preflight = require('./preflight');
@@ -286,19 +287,22 @@ async function guardSessions(ctx, svc, opts, record, when) {
@@ -286,19 +287,22 @@ async function guardSessions(ctx, svc, opts, record, when) {
let quiet = 0;
let quiet = 0;
let last = first;
let last = first;
log(`${describe(first)}; --wait-idle: holding the restart (${when}); need ${svc.drain.quietChecks} idle checks ${svc.drain.pollSeconds}s apart, max ${svc.drain.waitMaxSeconds}s`);
log(`${describe(first)}; --wait-idle: holding the restart (${when}); need ${svc.drain.quietChecks} idle checks ${svc.drain.pollSeconds}s apart, max ${svc.drain.waitMaxSeconds}s`);
- while (exec.now() - start < maxMs) {
- await exec.sleep(pollMs);
- last = await probes.countProtected(exec, svc);
- if (last.notRunning || last.count === 0) quiet += 1; else quiet = 0;
- if (quiet >= svc.drain.quietChecks) {
- record.protectedSessions.waitedSeconds = Math.round((exec.now() - start) / 1000);
- record.protectedSessions.count = 0;
- log(`idle for ${quiet} checks — proceeding (waited ${record.protectedSessions.waitedSeconds}s)`);
- return;
+ const poll = async () => {
+ while (exec.now() - start < maxMs) {
+ await exec.sleep(pollMs);
+ last = await probes.countProtected(exec, svc);
+ if (last.notRunning || last.count === 0) quiet += 1; else quiet = 0;
+ if (quiet >= svc.drain.quietChecks) {
+ record.protectedSessions.waitedSeconds = Math.round((exec.now() - start) / 1000);
+ record.protectedSessions.count = 0;
+ log(`idle for ${quiet} checks — proceeding (waited ${record.protectedSessions.waitedSeconds}s)`);
+ return;
+ }
+ log(`${describe(last)}; still holding (${Math.round((exec.now() - start) / 1000)}s)`);
}
}
- log(`${describe(last)}; still holding (${Math.round((exec.now() - start) / 1000)}s)`);
- }
- throw new OpError(`${describe(last)} after waiting ${svc.drain.waitMaxSeconds}s — gave up; nothing was restarted`, EXIT.PROTECTED, { result: 'gave-up-waiting' });
+ throw new OpError(`${describe(last)} after waiting ${svc.drain.waitMaxSeconds}s — gave up; nothing was restarted`, EXIT.PROTECTED, { result: 'gave-up-waiting' });
+ };
+ return opts.onIdleWait ? opts.onIdleWait(poll) : poll();
}
}
// ── helpers ──────────────────────────────────────────────────────────────────
// ── helpers ──────────────────────────────────────────────────────────────────
@@ -307,7 +311,7 @@ async function installAll(ctx, svc, g, installs, record) {
@@ -307,7 +311,7 @@ async function installAll(ctx, svc, g, installs, record) {
for (const i of installs) {
for (const i of installs) {
if (!i.install) continue;
if (!i.install) continue;
ctx.log(`installing dependencies in ${i.pkg} (${[i.lockfileChanged && 'lockfile changed', i.depsChanged && 'package.json dependencies changed', i.missingNodeModules && 'no node_modules', svc.install.always && 'always'].filter(Boolean).join(', ')})`);
ctx.log(`installing dependencies in ${i.pkg} (${[i.lockfileChanged && 'lockfile changed', i.depsChanged && 'package.json dependencies changed', i.missingNodeModules && 'no node_modules', svc.install.always && 'always'].filter(Boolean).join(', ')})`);
- await deps.install(ctx.exec, svc, g, i.pkg);
+ await deps.install(ctx.exec, svc, g, i.pkg, ctx.log);
record.installed.push(i.pkg);
record.installed.push(i.pkg);
}
}
}
}
@@ -395,7 +399,7 @@ async function restoreCheckout(ctx, svc, g, sha, record) {
@@ -395,7 +399,7 @@ async function restoreCheckout(ctx, svc, g, sha, record) {
if (record.installed.length) {
if (record.installed.length) {
for (const pkg of record.installed) {
for (const pkg of record.installed) {
ctx.log(`reinstalling dependencies in ${pkg} for ${sha.slice(0, 12)}`);
ctx.log(`reinstalling dependencies in ${pkg} for ${sha.slice(0, 12)}`);
- await deps.install(ctx.exec, svc, g, pkg);
+ await deps.install(ctx.exec, svc, g, pkg, ctx.log);
}
}
}
}
const v = await verifyDeps(ctx, svc, g);
const v = await verifyDeps(ctx, svc, g);
@@ -440,6 +444,8 @@ async function apply(ctx, svc, mode, opts = {}) {
@@ -440,6 +444,8 @@ async function apply(ctx, svc, mode, opts = {}) {
}
}
const held = await lock.acquire(exec, inv, svc.id);
const held = await lock.acquire(exec, inv, svc.id);
if (held.recoveredStale) log(`removed a stale lock (${held.file})`);
if (held.recoveredStale) log(`removed a stale lock (${held.file})`);
+ const hostLock = { current: null };
+ const guardedOpts = { ...opts, onIdleWait: (work) => deployGuard.waitOutsideLock(ctx, svc, hostLock, work) };
const g = git(exec, svc);
const g = git(exec, svc);
const record = {
const record = {
id: releases.newId(exec.now()),
id: releases.newId(exec.now()),
@@ -464,6 +470,7 @@ async function apply(ctx, svc, mode, opts = {}) {
@@ -464,6 +470,7 @@ async function apply(ctx, svc, mode, opts = {}) {
let exitCode = EXIT.OK;
let exitCode = EXIT.OK;
let writeRecord = true;
let writeRecord = true;
try {
try {
+ hostLock.current = await deployGuard.acquire(ctx, svc);
const plan = await computePlan(ctx, svc, { mode, to: opts.to, fetch: opts.fetch !== false, restart: opts.restart });
const plan = await computePlan(ctx, svc, { mode, to: opts.to, fetch: opts.fetch !== false, restart: opts.restart });
Object.assign(record, { from: plan.from, to: plan.to, commits: plan.commits.length, lockfileChanged: plan.lockfileChanged, restartNeeded: plan.restartNeeded });
Object.assign(record, { from: plan.from, to: plan.to, commits: plan.commits.length, lockfileChanged: plan.lockfileChanged, restartNeeded: plan.restartNeeded });
log(`${svc.id}: ${mode} ${plan.from.slice(0, 12)} → ${plan.to.slice(0, 12)} (${plan.commits.length} commit(s), ${plan.changedFiles} file(s) changed)`);
log(`${svc.id}: ${mode} ${plan.from.slice(0, 12)} → ${plan.to.slice(0, 12)} (${plan.commits.length} commit(s), ${plan.changedFiles} file(s) changed)`);
@@ -487,7 +494,7 @@ async function apply(ctx, svc, mode, opts = {}) {
@@ -487,7 +494,7 @@ async function apply(ctx, svc, mode, opts = {}) {
if (plan.unitDrift.length && !opts.installUnits && !svc.installUnits) log(`note: unit files differ from the repo (${plan.unitDrift.map((u) => u.unit).join(', ')}); pass --install-units to install them`);
if (plan.unitDrift.length && !opts.installUnits && !svc.installUnits) log(`note: unit files differ from the repo (${plan.unitDrift.map((u) => u.unit).join(', ')}); pass --install-units to install them`);
// Sessions first: in place, the running process would otherwise see its files change under it.
// Sessions first: in place, the running process would otherwise see its files change under it.
- if (plan.restartNeeded) await guardSessions(ctx, svc, opts, record, 'before the checkout moves');
+ if (plan.restartNeeded) await guardSessions(ctx, svc, guardedOpts, record, 'before the checkout moves');
if (plan.generatedDirty.length && !plan.upToDate) {
if (plan.generatedDirty.length && !plan.upToDate) {
log(`restoring ${plan.generatedDirty.length} tracked build output file(s) from git (${svc.generated.join(', ')})`);
log(`restoring ${plan.generatedDirty.length} tracked build output file(s) from git (${svc.generated.join(', ')})`);
@@ -538,7 +545,7 @@ async function apply(ctx, svc, mode, opts = {}) {
@@ -538,7 +545,7 @@ async function apply(ctx, svc, mode, opts = {}) {
}
}
if (opts.installUnits || svc.installUnits) await installUnits(ctx, svc, record);
if (opts.installUnits || svc.installUnits) await installUnits(ctx, svc, record);
if (plan.backupNeeded && plan.restartNeeded) await backupBeforeRestart(ctx, svc, record);
if (plan.backupNeeded && plan.restartNeeded) await backupBeforeRestart(ctx, svc, record);
- if (plan.restartNeeded) await guardSessions(ctx, svc, opts, record, 'immediately before the restart');
+ if (plan.restartNeeded) await guardSessions(ctx, svc, guardedOpts, record, 'immediately before the restart');
} catch (err) {
} catch (err) {
// Nothing has been restarted: put the files back the way the running process expects.
// Nothing has been restarted: put the files back the way the running process expects.
if (record.checkoutMoved) {
if (record.checkoutMoved) {
@@ -606,6 +613,7 @@ async function apply(ctx, svc, mode, opts = {}) {
@@ -606,6 +613,7 @@ async function apply(ctx, svc, mode, opts = {}) {
if (writeRecord) {
if (writeRecord) {
try { await releases.append(exec, inv, record); } catch (e) { log(`warning: could not append to the release log: ${e.message}`); }
try { await releases.append(exec, inv, record); } catch (e) { log(`warning: could not append to the release log: ${e.message}`); }
}
}
+ if (hostLock.current) await hostLock.current.release();
await held.release();
await held.release();
}
}
}
}
new file mode 100644
new file mode 100644
@@ -0,0 +1,108 @@
@@ -0,0 +1,108 @@
+'use strict';
+const assert = require('assert');
+const { scenario, push, test, runTests } = require('./helpers');
+const { liveHost } = require('./strategy-hosts');
+
+const tick = () => new Promise((resolve) => setImmediate(resolve));
+
+runTests([
+ test('concurrent deploys serialize preparation across services', async () => {
+ const host = scenario();
+ push(host, 'live', { 'package-lock.json': '{"lockfileVersion":3,"v":2}', 'server/index.js': 'v2();' });
+ const oldMedia = host.repo('media').head;
+ push(host, 'media', { 'server/index.js': 'v2();' });
+ const originalRun = host.exec.run.bind(host.exec);
+ let entered;
+ const installing = new Promise((resolve) => { entered = resolve; });
+ let continueInstall;
+ const barrier = new Promise((resolve) => { continueInstall = resolve; });
+ host.exec.run = async (cmd, args, opts) => {
+ if (cmd === 'npm' && !host.installPaused) {
+ host.installPaused = true;
+ entered();
+ await barrier;
+ }
+ return originalRun(cmd, args, opts);
+ };
+ const first = host.cli('deploy', 'live');
+ await installing;
+ const second = host.cli('deploy', 'media');
+ await tick();
+ assert.deepStrictEqual(host.deployLockEvents.slice(0, 2), ['acquire:live', 'wait:media:live']);
+ assert.strictEqual(host.repo('media').head, oldMedia, 'media did not prepare while live installed');
+ continueInstall();
+ const [a, b] = await Promise.all([first, second]);
+ assert.strictEqual(a.code, 0, a.out);
+ assert.strictEqual(b.code, 0, b.out);
+ assert.match(b.out, /waiting for live's deploy to finish/);
+ assert.deepStrictEqual(host.deployLockEvents, ['acquire:live', 'wait:media:live', 'release:live', 'acquire:media', 'release:media']);
+ }),
+
+ test('--wait-idle releases the host lock while streams are active', async () => {
+ const host = scenario();
+ push(host, 'live', { 'server/index.js': 'v2();' });
+ push(host, 'media', { 'server/index.js': 'v2();' });
+ let probes = 0;
+ host.http.set('http://127.0.0.1:3000/api/streams', () => ({ status: 200, body: { streams: ++probes <= 2 ? [{ is_live: true }] : [] } }));
+ const originalSleep = host.exec.sleep.bind(host.exec);
+ let sleeping;
+ const entered = new Promise((resolve) => { sleeping = resolve; });
+ let resume;
+ const barrier = new Promise((resolve) => { resume = resolve; });
+ host.exec.sleep = async (ms) => {
+ if (ms === 60000 && !host.waitPaused) { host.waitPaused = true; sleeping(); await barrier; }
+ return originalSleep(ms);
+ };
+ const first = host.cli('deploy', 'live', '--wait-idle');
+ await entered;
+ assert.deepStrictEqual(host.deployLockEvents.slice(0, 2), ['acquire:live', 'release:live']);
+ const other = await host.cli('deploy', 'media');
+ assert.strictEqual(other.code, 0, other.out);
+ resume();
+ const result = await first;
+ assert.strictEqual(result.code, 0, result.out);
+ assert.deepStrictEqual(host.deployLockEvents, ['acquire:live', 'release:live', 'acquire:media', 'release:media', 'acquire:live', 'release:live']);
+ }),
+
+ test('memory guard waits for MemAvailable and then installs', async () => {
+ const host = scenario();
+ push(host, 'live', { 'package-lock.json': '{"lockfileVersion":3,"v":2}' });
+ const readings = [512, 768, 2048];
+ host.memAvailableMiB = () => readings.shift() || 2048;
+ const r = await host.cli('deploy', 'live');
+ assert.strictEqual(r.code, 0, r.out);
+ assert.match(r.out, /MemAvailable is 512 MiB, below 1536 MiB/);
+ assert.match(r.out, /MemAvailable is 2048 MiB; proceeding/);
+ assert.ok(host.calls.some((c) => c.cmd === 'npm'));
+ }),
+
+ test('memory guard times out without starting npm or restarting', async () => {
+ const host = scenario();
+ const from = host.repo('live').head;
+ push(host, 'live', { 'package-lock.json': '{"lockfileVersion":3,"v":2}' });
+ host.memAvailableMiB = 512;
+ const inventory = JSON.parse(host.read('/etc/openvibe/host.json'));
+ inventory.deployGuard = { memoryWaitSeconds: 10 };
+ host.put('/etc/openvibe/host.json', JSON.stringify(inventory), { mode: 0o640, owner: 'root' });
+ const r = await host.cli('deploy', 'live');
+ assert.strictEqual(r.code, 2, r.out);
+ assert.match(r.out, /MemAvailable stayed below 1536 MiB for 10s/);
+ assert.strictEqual(host.repo('live').head, from);
+ assert.ok(!host.calls.some((c) => c.cmd === 'npm'));
+ assert.deepStrictEqual(host.restarts(), []);
+ }),
+ test('release-layout memory timeout removes the prepared release', async () => {
+ const host = await liveHost();
+ host.push({ 'package-lock.json': '{"lockfileVersion":3,"v":2}' }, 'dependencies');
+ host.memAvailableMiB = 512;
+ const inventory = JSON.parse(host.read('/etc/openvibe/host.json'));
+ inventory.deployGuard = { memoryWaitSeconds: 10 };
+ host.put('/etc/openvibe/host.json', JSON.stringify(inventory), { mode: 0o640, owner: 'root' });
+ const result = await host.cli('deploy', 'live');
+ assert.strictEqual(result.code, 2, result.out);
+ assert.deepStrictEqual(await host.releaseIds(), [host.firstRelease]);
+ assert.strictEqual(host.current(), host.firstRelease);
+ assert.ok(!host.calls.some((c) => c.cmd === 'npm'));
+ assert.deepStrictEqual(host.restarts(), []);
+ }),
+]);
@@ -15,6 +15,20 @@ const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'ovhost-exec-'));
@@ -15,6 +15,20 @@ const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'ovhost-exec-'));
process.on('exit', () => fs.rmSync(dir, { recursive: true, force: true }));
process.on('exit', () => fs.rmSync(dir, { recursive: true, force: true }));
runTests([
runTests([
+ test('kernel flock serializes two holders and reports the service that owns it', async () => {
+ const file = path.join(dir, 'deploy.lock');
+ const first = await exec.acquireDeployLock(file, 'live', 2);
+ let waiting = null;
+ const secondPromise = exec.acquireDeployLock(file, 'media', 2, (owner) => { waiting = owner; });
+ await new Promise((resolve) => setTimeout(resolve, 350));
+ assert.strictEqual(waiting, 'live');
+ await assert.rejects(exec.acquireDeployLock(file, 'third', 0.1), /timed out waiting for the host deploy lock/);
+ await first.release();
+ const second = await secondPromise;
+ assert.strictEqual(fs.readFileSync(file, 'utf8').trim(), 'media');
+ await second.release();
+ assert.ok((await exec.memAvailableBytes()) > 0);
+ }),
test('http, files, locks and run() behave as the fake host assumes', async () => {
test('http, files, locks and run() behave as the fake host assumes', async () => {
const server = http.createServer((req, res) => { res.writeHead(req.url === '/ok' ? 200 : 503); res.end(JSON.stringify({ host: req.headers.host })); });
const server = http.createServer((req, res) => { res.writeHead(req.url === '/ok' ? 200 : 503); res.end(JSON.stringify({ host: req.headers.host })); });
await new Promise((r) => server.listen(0, '127.0.0.1', r));
await new Promise((r) => server.listen(0, '127.0.0.1', r));
@@ -24,10 +24,14 @@ function createFakeHost({ root = true, user = 'root', hostname = 'fake-host', st
@@ -24,10 +24,14 @@ function createFakeHost({ root = true, user = 'root', hostname = 'fake-host', st
const http = new Map(); // url -> fn({ headers, method, body }) -> { status, body } (request() also passes method and body)
const http = new Map(); // url -> fn({ headers, method, body }) -> { status, body } (request() also passes method and body)
const alivePids = new Set();
const alivePids = new Set();
const listeners = new Map(); // port -> [{ pid, process }]
const listeners = new Map(); // port -> [{ pid, process }]
+ const deployLockQueue = [];
+ let deployLockOwner = null;
let clock = start;
let clock = start;
let seq = 0;
let seq = 0;
const host = {
const host = {
files, calls, repos, remoteRepos, units, http, alivePids, listeners,
files, calls, repos, remoteRepos, units, http, alivePids, listeners,
+ deployLockEvents: [],
+ memAvailableMiB: 4096,
nginxTest: () => ({ code: 0, stderr: 'nginx: configuration file /etc/nginx/nginx.conf test is successful' }),
nginxTest: () => ({ code: 0, stderr: 'nginx: configuration file /etc/nginx/nginx.conf test is successful' }),
npm: null, // (cwd, argv, as) -> undefined | { code, stderr } ; default installs package.json deps
npm: null, // (cwd, argv, as) -> undefined | { code, stderr } ; default installs package.json deps
npmRewritesLockfile: true,
npmRewritesLockfile: true,
@@ -765,6 +769,21 @@ function createFakeHost({ root = true, user = 'root', hostname = 'fake-host', st
@@ -765,6 +769,21 @@ function createFakeHost({ root = true, user = 'root', hostname = 'fake-host', st
}
}
return true;
return true;
},
},
+ async acquireDeployLock(file, service, timeoutSeconds, onWait) {
+ if (deployLockOwner) {
+ host.deployLockEvents.push(`wait:${service}:${deployLockOwner}`);
+ onWait(deployLockOwner);
+ await new Promise((resolve) => deployLockQueue.push(resolve));
+ }
+ deployLockOwner = service;
+ host.deployLockEvents.push(`acquire:${service}`);
+ return { release: async () => {
+ host.deployLockEvents.push(`release:${service}`);
+ deployLockOwner = null;
+ if (deployLockQueue.length) deployLockQueue.shift()();
+ } };
+ },
+ async memAvailableBytes() { return (typeof host.memAvailableMiB === 'function' ? host.memAvailableMiB() : host.memAvailableMiB) * 1024 * 1024; },
async sleep(ms) { clock += ms; },
async sleep(ms) { clock += ms; },
now: () => clock,
now: () => clock,
async isRoot() { return root; },
async isRoot() { return root; },