openvibers/openvibe.host Commits
Release layout: pruning keeps the release the service's process runs from (its /proc cwd); static-only deploys move current without a restart, and on 10-17 Live ran 2.5 h from a pruned release, so every /shared script (the navbar), live thumbnail and AI frame failed with ENOENT (#131)
Changes
@@@ -22,8 +22,9 @@* paths inside the release), restart the units, wait for readiness (and /release.json naming the* new sha when ready.release). Not ready: switch back, put the previous release's unit files back,* restart, exit 3; if that is not ready either, exit 4 (MANUAL INTERVENTION).-- * 5. prune to release.keep, never the current release, the one just left, or one a worker unit-- * instance (OpenRestream's openre-rtmp-ingest@<id>) still runs from.++ * 5. prune to release.keep, never the current release, the one just left, the one the service's++ * process runs from (its cwd: static-only deploys leave it behind), or one a worker unit instance++ * (OpenRestream's openre-rtmp-ingest@<id>) still runs from.** The socket rule (Live): pid 1 must hold the socket unit's listener. When the socket unit file changed* in this deploy, or systemd does not hold the listener, the service is stopped, the socket restarted@@@ -102,6 +103,18 @@ async function listReleases(ctx, svc) {/** Release ids a worker unit instance still runs from (openre-rtmp-ingest@<id>.service). */async function releasesInUse(ctx, svc) {const used = new Set();++ // The service's own process: a static-only deploy switches `current` without a restart, so the process++ // can still run from a release several deploys back. Its cwd is that release (systemd resolves `current`++ // when it starts the unit). 2026-10-17: Live ran from a pruned release for 2.5 hours; every /shared file,++ // live thumbnail and AI frame it read or wrote under its own release failed with ENOENT.++ for (const u of svc.units) {++ let pid = 0;++ try { pid = (await systemd.show(ctx.exec, u)).mainPid; } catch { continue; }++ if (!pid) continue;++ const cwd = await ctx.exec.readlink(`/proc/${pid}/cwd`);++ const rel = cwd && path.relative(svc.release.releasesDir, cwd);++ if (rel && !rel.startsWith('..') && !path.isAbsolute(rel)) used.add(rel.split(path.sep)[0]);++ }for (const w of svc.workerUnits) {if (!w.endsWith('@.service')) continue;let list = [];@@@ -329,7 +342,7 @@ async function restartRelease(ctx, svc, { socketChanged = false, expectSha = nulreturn { ...r, rebind, socketHeld };}--/** Remove releases beyond keep: never current, the one just left, or one a worker still runs from. */++/** Remove releases beyond keep: never current, the one just left, or one the service or a worker still runs from. */async function prune(ctx, svc, { keepIds = [] } = {}) {const list = await listReleases(ctx, svc);const cur = await currentRelease(ctx, svc);@@@ -59,6 +59,7 @@ function createFakeHost({ root = true, user = 'root', hostname = 'fake-host', stjournalctl: null, // (args, opts) -> { code, stdout, stderr } ; default: no entriesjournalctlCalls: [], // { args, as, privileged }onRestart: null, // (unit) -> void++ procCwd: new Map(), // pid -> the directory /proc/<pid>/cwd names (a process's working directory)onSystemdRun: null, // (spec) -> undefined | { code, stderr } ; spec = { unit, uid, cwd, props, argv, pid }onKill: null, // (pid, signal) -> false to ignore the signalonNode: null, // (args, opts) -> undefined | { code, stdout, stderr } ; `node` otherwise exits 0@@@ -691,7 +692,11 @@ function createFakeHost({ root = true, user = 'root', hostname = 'fake-host', stif (files.has(lpath(p))) throw Object.assign(new Error(`EEXIST: file already exists, symlink '${target}' -> '${p}'`), { code: 'EEXIST' });ensureDir(path.dirname(p)); files.set(lpath(p), { type: 'symlink', target, owner: user, mode: 0o777, mtime: clock });},-- async readlink(p) { const e = files.get(lpath(p)); return e && e.type === 'symlink' ? e.target : null; },++ async readlink(p) {++ const proc = /^\/proc\/(\d+)\/cwd$/.exec(p);++ if (proc) return host.procCwd.get(Number(proc[1])) || null;++ const e = files.get(lpath(p)); return e && e.type === 'symlink' ? e.target : null;++ },async stat(p) {const l = files.get(lpath(p));const e = get(p);@@@ -100,7 +100,7 @@ async function releaseLayoutHost({ id, entry, files, initialId, unitFiles = {},await host.exec.symlink(first, svc.release.current);for (const [unit, src] of Object.entries(svc.unitSources)) if (files[src] != null) host.put(path.join('/etc/systemd/system', unit), files[src]);for (const [unit, text] of Object.entries(unitFiles)) host.put(path.join('/etc/systemd/system', unit), text);-- for (const u of svc.units) host.addUnit(u, { runningSha: sha });++ for (const u of svc.units) host.procCwd.set(host.addUnit(u, { runningSha: sha }).mainPid, first);host.firstRelease = path.basename(first);host.releaseDir = (rid) => path.join(svc.release.releasesDir, rid);host.current = () => path.basename(host.resolve(svc.release.current));@@@ -113,6 +113,7 @@ async function releaseLayoutHost({ id, entry, files, initialId, unitFiles = {},if (unit.endsWith('.socket')) { if (host.onSocketRestart) host.onSocketRestart(unit); return; }const wt = host.repos.get(host.resolve(svc.release.current));u.runningSha = wt ? wt.head : null;++ host.procCwd.set(u.mainPid, host.resolve(svc.release.current));};host.readyFor = (unit) => () => {const u = host.units.get(unit);@@@ -303,6 +303,31 @@ runTests([assert.ok(host.calls.some((c) => c.cmd === 'git' && c.args.includes('remove')), 'pruned through git worktree remove');}),++ test('live: the release the running process started from survives pruning while static deploys move current on', async () => {++ const host = await liveHost();++ host.push({ 'server/index.js': 'v2();' }, 'server');++ assert.strictEqual((await host.cli('deploy', 'live')).code, 0);++ const running = host.current();++ // Four static-only deploys: `current` moves four times, the process keeps running from `running`.++ for (let i = 1; i <= 4; i++) {++ host.exec.sleep(1000);++ host.push({ 'public/app.js': `app${i}();` }, `static ${i}`);++ const r = await host.cli('deploy', 'live');++ assert.strictEqual(r.code, 0, r.out);++ assert.match(r.out, /without a restart/);++ }++ assert.deepStrictEqual(host.restarts(), ['openvibe-live.service'], 'one restart, for the server change');++ const ids = await host.releaseIds();++ assert.ok(ids.includes(running), `the running process's release is kept (${ids.join(' ')})`);++ assert.ok(ids.includes(host.current()));++ assert.strictEqual(ids.length, 4, 'keep (3) plus the one the process runs from');++ // Once the process restarts onto `current`, its old release is pruned like any other.++ host.exec.sleep(1000);++ host.push({ 'server/index.js': 'v3();' }, 'server again');++ assert.strictEqual((await host.cli('deploy', 'live')).code, 0);++ assert.ok(!(await host.releaseIds()).includes(running));++ }),++test('live: the entry may name the current link as its repo; the release root is its parent', async () => {const { normalise } = require('../lib/inventory');const inv = normalise({ services: { live: { repo: '/opt/openvibe.live/current', owner: 'root', strategy: 'release-layout', units: ['openvibe-live.service'] } } });
@@ -22,8 +22,9 @@
@@ -22,8 +22,9 @@
* paths inside the release), restart the units, wait for readiness (and /release.json naming the
* paths inside the release), restart the units, wait for readiness (and /release.json naming the
* new sha when ready.release). Not ready: switch back, put the previous release's unit files back,
* new sha when ready.release). Not ready: switch back, put the previous release's unit files back,
* restart, exit 3; if that is not ready either, exit 4 (MANUAL INTERVENTION).
* restart, exit 3; if that is not ready either, exit 4 (MANUAL INTERVENTION).
- * 5. prune to release.keep, never the current release, the one just left, or one a worker unit
- * instance (OpenRestream's openre-rtmp-ingest@<id>) still runs from.
+ * 5. prune to release.keep, never the current release, the one just left, the one the service's
+ * process runs from (its cwd: static-only deploys leave it behind), or one a worker unit instance
+ * (OpenRestream's openre-rtmp-ingest@<id>) still runs from.
*
*
* The socket rule (Live): pid 1 must hold the socket unit's listener. When the socket unit file changed
* The socket rule (Live): pid 1 must hold the socket unit's listener. When the socket unit file changed
* in this deploy, or systemd does not hold the listener, the service is stopped, the socket restarted
* in this deploy, or systemd does not hold the listener, the service is stopped, the socket restarted
@@ -102,6 +103,18 @@ async function listReleases(ctx, svc) {
@@ -102,6 +103,18 @@ async function listReleases(ctx, svc) {
/** Release ids a worker unit instance still runs from (openre-rtmp-ingest@<id>.service). */
/** Release ids a worker unit instance still runs from (openre-rtmp-ingest@<id>.service). */
async function releasesInUse(ctx, svc) {
async function releasesInUse(ctx, svc) {
const used = new Set();
const used = new Set();
+ // The service's own process: a static-only deploy switches `current` without a restart, so the process
+ // can still run from a release several deploys back. Its cwd is that release (systemd resolves `current`
+ // when it starts the unit). 2026-10-17: Live ran from a pruned release for 2.5 hours; every /shared file,
+ // live thumbnail and AI frame it read or wrote under its own release failed with ENOENT.
+ for (const u of svc.units) {
+ let pid = 0;
+ try { pid = (await systemd.show(ctx.exec, u)).mainPid; } catch { continue; }
+ if (!pid) continue;
+ const cwd = await ctx.exec.readlink(`/proc/${pid}/cwd`);
+ const rel = cwd && path.relative(svc.release.releasesDir, cwd);
+ if (rel && !rel.startsWith('..') && !path.isAbsolute(rel)) used.add(rel.split(path.sep)[0]);
+ }
for (const w of svc.workerUnits) {
for (const w of svc.workerUnits) {
if (!w.endsWith('@.service')) continue;
if (!w.endsWith('@.service')) continue;
let list = [];
let list = [];
@@ -329,7 +342,7 @@ async function restartRelease(ctx, svc, { socketChanged = false, expectSha = nul
@@ -329,7 +342,7 @@ async function restartRelease(ctx, svc, { socketChanged = false, expectSha = nul
return { ...r, rebind, socketHeld };
return { ...r, rebind, socketHeld };
}
}
-/** Remove releases beyond keep: never current, the one just left, or one a worker still runs from. */
+/** Remove releases beyond keep: never current, the one just left, or one the service or a worker still runs from. */
async function prune(ctx, svc, { keepIds = [] } = {}) {
async function prune(ctx, svc, { keepIds = [] } = {}) {
const list = await listReleases(ctx, svc);
const list = await listReleases(ctx, svc);
const cur = await currentRelease(ctx, svc);
const cur = await currentRelease(ctx, svc);
@@ -59,6 +59,7 @@ function createFakeHost({ root = true, user = 'root', hostname = 'fake-host', st
@@ -59,6 +59,7 @@ function createFakeHost({ root = true, user = 'root', hostname = 'fake-host', st
journalctl: null, // (args, opts) -> { code, stdout, stderr } ; default: no entries
journalctl: null, // (args, opts) -> { code, stdout, stderr } ; default: no entries
journalctlCalls: [], // { args, as, privileged }
journalctlCalls: [], // { args, as, privileged }
onRestart: null, // (unit) -> void
onRestart: null, // (unit) -> void
+ procCwd: new Map(), // pid -> the directory /proc/<pid>/cwd names (a process's working directory)
onSystemdRun: null, // (spec) -> undefined | { code, stderr } ; spec = { unit, uid, cwd, props, argv, pid }
onSystemdRun: null, // (spec) -> undefined | { code, stderr } ; spec = { unit, uid, cwd, props, argv, pid }
onKill: null, // (pid, signal) -> false to ignore the signal
onKill: null, // (pid, signal) -> false to ignore the signal
onNode: null, // (args, opts) -> undefined | { code, stdout, stderr } ; `node` otherwise exits 0
onNode: null, // (args, opts) -> undefined | { code, stdout, stderr } ; `node` otherwise exits 0
@@ -691,7 +692,11 @@ function createFakeHost({ root = true, user = 'root', hostname = 'fake-host', st
@@ -691,7 +692,11 @@ function createFakeHost({ root = true, user = 'root', hostname = 'fake-host', st
if (files.has(lpath(p))) throw Object.assign(new Error(`EEXIST: file already exists, symlink '${target}' -> '${p}'`), { code: 'EEXIST' });
if (files.has(lpath(p))) throw Object.assign(new Error(`EEXIST: file already exists, symlink '${target}' -> '${p}'`), { code: 'EEXIST' });
ensureDir(path.dirname(p)); files.set(lpath(p), { type: 'symlink', target, owner: user, mode: 0o777, mtime: clock });
ensureDir(path.dirname(p)); files.set(lpath(p), { type: 'symlink', target, owner: user, mode: 0o777, mtime: clock });
},
},
- async readlink(p) { const e = files.get(lpath(p)); return e && e.type === 'symlink' ? e.target : null; },
+ async readlink(p) {
+ const proc = /^\/proc\/(\d+)\/cwd$/.exec(p);
+ if (proc) return host.procCwd.get(Number(proc[1])) || null;
+ const e = files.get(lpath(p)); return e && e.type === 'symlink' ? e.target : null;
+ },
async stat(p) {
async stat(p) {
const l = files.get(lpath(p));
const l = files.get(lpath(p));
const e = get(p);
const e = get(p);
@@ -100,7 +100,7 @@ async function releaseLayoutHost({ id, entry, files, initialId, unitFiles = {},
@@ -100,7 +100,7 @@ async function releaseLayoutHost({ id, entry, files, initialId, unitFiles = {},
await host.exec.symlink(first, svc.release.current);
await host.exec.symlink(first, svc.release.current);
for (const [unit, src] of Object.entries(svc.unitSources)) if (files[src] != null) host.put(path.join('/etc/systemd/system', unit), files[src]);
for (const [unit, src] of Object.entries(svc.unitSources)) if (files[src] != null) host.put(path.join('/etc/systemd/system', unit), files[src]);
for (const [unit, text] of Object.entries(unitFiles)) host.put(path.join('/etc/systemd/system', unit), text);
for (const [unit, text] of Object.entries(unitFiles)) host.put(path.join('/etc/systemd/system', unit), text);
- for (const u of svc.units) host.addUnit(u, { runningSha: sha });
+ for (const u of svc.units) host.procCwd.set(host.addUnit(u, { runningSha: sha }).mainPid, first);
host.firstRelease = path.basename(first);
host.firstRelease = path.basename(first);
host.releaseDir = (rid) => path.join(svc.release.releasesDir, rid);
host.releaseDir = (rid) => path.join(svc.release.releasesDir, rid);
host.current = () => path.basename(host.resolve(svc.release.current));
host.current = () => path.basename(host.resolve(svc.release.current));
@@ -113,6 +113,7 @@ async function releaseLayoutHost({ id, entry, files, initialId, unitFiles = {},
@@ -113,6 +113,7 @@ async function releaseLayoutHost({ id, entry, files, initialId, unitFiles = {},
if (unit.endsWith('.socket')) { if (host.onSocketRestart) host.onSocketRestart(unit); return; }
if (unit.endsWith('.socket')) { if (host.onSocketRestart) host.onSocketRestart(unit); return; }
const wt = host.repos.get(host.resolve(svc.release.current));
const wt = host.repos.get(host.resolve(svc.release.current));
u.runningSha = wt ? wt.head : null;
u.runningSha = wt ? wt.head : null;
+ host.procCwd.set(u.mainPid, host.resolve(svc.release.current));
};
};
host.readyFor = (unit) => () => {
host.readyFor = (unit) => () => {
const u = host.units.get(unit);
const u = host.units.get(unit);
@@ -303,6 +303,31 @@ runTests([
@@ -303,6 +303,31 @@ runTests([
assert.ok(host.calls.some((c) => c.cmd === 'git' && c.args.includes('remove')), 'pruned through git worktree remove');
assert.ok(host.calls.some((c) => c.cmd === 'git' && c.args.includes('remove')), 'pruned through git worktree remove');
}),
}),
+ test('live: the release the running process started from survives pruning while static deploys move current on', async () => {
+ const host = await liveHost();
+ host.push({ 'server/index.js': 'v2();' }, 'server');
+ assert.strictEqual((await host.cli('deploy', 'live')).code, 0);
+ const running = host.current();
+ // Four static-only deploys: `current` moves four times, the process keeps running from `running`.
+ for (let i = 1; i <= 4; i++) {
+ host.exec.sleep(1000);
+ host.push({ 'public/app.js': `app${i}();` }, `static ${i}`);
+ const r = await host.cli('deploy', 'live');
+ assert.strictEqual(r.code, 0, r.out);
+ assert.match(r.out, /without a restart/);
+ }
+ assert.deepStrictEqual(host.restarts(), ['openvibe-live.service'], 'one restart, for the server change');
+ const ids = await host.releaseIds();
+ assert.ok(ids.includes(running), `the running process's release is kept (${ids.join(' ')})`);
+ assert.ok(ids.includes(host.current()));
+ assert.strictEqual(ids.length, 4, 'keep (3) plus the one the process runs from');
+ // Once the process restarts onto `current`, its old release is pruned like any other.
+ host.exec.sleep(1000);
+ host.push({ 'server/index.js': 'v3();' }, 'server again');
+ assert.strictEqual((await host.cli('deploy', 'live')).code, 0);
+ assert.ok(!(await host.releaseIds()).includes(running));
+ }),
+
test('live: the entry may name the current link as its repo; the release root is its parent', async () => {
test('live: the entry may name the current link as its repo; the release root is its parent', async () => {
const { normalise } = require('../lib/inventory');
const { normalise } = require('../lib/inventory');
const inv = normalise({ services: { live: { repo: '/opt/openvibe.live/current', owner: 'root', strategy: 'release-layout', units: ['openvibe-live.service'] } } });
const inv = normalise({ services: { live: { repo: '/opt/openvibe.live/current', owner: 'root', strategy: 'release-layout', units: ['openvibe-live.service'] } } });