openvibers/openvibe.media Commits
Recordings end when ffmpeg ends again (broken by the PostgreSQL switch this morning), and avatar ingest awaits its slug.
Changes
@@@ -103,7 +103,7 @@ function createIngestHandler({ db, config, screenshotsDir, generateSlug, log = cfs.mkdirSync(screenshotsDir, { recursive: true });const file = path.join(screenshotsDir, `avatar-n${parseInt(user_id, 10) || 0}-${Date.now()}-${crypto.randomBytes(4).toString('hex')}.webp`);fs.writeFileSync(file, pic.buffer);-- const slug = generateSlug();++ const slug = await generateSlug(); // async since the PostgreSQL switch (it checks the table)const insert = async () => await db.run(`INSERT INTO pastes (app_id, slug, user_id, type, title, content, language, visibility, screenshot_path, metadata, ip_address)VALUES ('network', ?, ?, 'screenshot', ?, '', 'text', 'unlisted', ?, ?, NULL) RETURNING id`,[slug, parseInt(user_id, 10) || null, `Avatar${username ? ' of ' + String(username).slice(0, 40) : ''}`, file,@@@ -333,7 +333,7 @@ class StreamRecorder {* Exactly the predecessor's RTMP path — no re-encode, always real-time,* live-seekable, kill-tolerant, its own lossless master.*/-- startRtmp(vod, rtmpUrl) {++ async startRtmp(vod, rtmpUrl) {const guard = this._guardCanRecord(vod);if (!guard.ok) return guard;// SSRF guard: ffmpeg connects only to an allow-listed RTMP server (config.rtmpPull.allow).@@@ -368,8 +368,12 @@ class StreamRecorder {return { ok: false, error: `ffmpeg spawn failed: ${err.message}` };}-- const recording = this._registerCommon(vod, proc, filePath, null, { protocol: 'rtmp' });-- this._wireProcess(recording, 'rtmp');++ // Listeners first, then the row: _registerCommon puts the recording in the map before its first await,++ // so ffmpeg's exit is caught even if it comes while the row is being written. (Unawaited, this left every++ // recording "in progress" forever after ffmpeg ended, and nothing finalized it.)++ const persisted = this._registerCommon(vod, proc, filePath, null, { protocol: 'rtmp' });++ this._wireProcess(this.activeRecordings.get(vod.id), 'rtmp');++ await persisted;console.log(`[VOD] Recording started: vod ${vod.id} → ${filename} (rtmp pull)`);return { ok: true, filePath };@@@ -384,7 +388,7 @@ class StreamRecorder {* @param {object} [audio] same shape (codec e.g. 'opus')* @returns {ok, videoPort, audioPort} — caller points PlainRtpTransports at 127.0.0.1*/-- startRtp(vod, video, audio) {++ async startRtp(vod, video, audio) {const guard = this._guardCanRecord(vod);if (!guard.ok) return guard;if (!video || video.payloadType == null || !video.codec) {@@@ -514,13 +518,14 @@ class StreamRecorder {return { ok: false, error: `ffmpeg spawn failed: ${err.message}` };}-- const recording = this._registerCommon(vod, proc, filePath, masterPath, {++ const persisted = this._registerCommon(vod, proc, filePath, masterPath, {protocol: 'rtp',sdpPath,rtpVideoPort: videoPort,rtpAudioPort: audioPort,});-- this._wireProcess(recording, 'rtp');++ this._wireProcess(this.activeRecordings.get(vod.id), 'rtp'); // before the row write (see startRtmp)++ await persisted;const recMode = passthrough ? `passthrough ${isH264 ? 'H.264→mp4' : 'VP8/9→webm'} copy` : 'libvpx re-encode';console.log(`[VOD] RTP recording started: vod ${vod.id} → ${filename} [${recMode}] (video:${videoPort}${audioPort ? ` audio:${audioPort}` : ''})`);@@@ -122,7 +122,7 @@ router.post('/:id/ingest/rtmp', tenantAuth(), async (req, res) => {try {const vod = await _getVodForWrite(req, res);if (!vod) return;-- const result = recorder.startRtmp(vod, req.body?.rtmp_url);++ const result = await recorder.startRtmp(vod, req.body?.rtmp_url);if (!result.ok) return res.status(result.status || 409).json({ error: result.error });res.status(202).json({ id: vod.id, status: 'recording' });} catch (err) {@@@ -137,7 +137,7 @@ router.post('/:id/ingest/rtp/start', tenantAuth(), async (req, res) => {const vod = await _getVodForWrite(req, res);if (!vod) return;const { video, audio } = req.body || {};-- const result = recorder.startRtp(vod, video, audio);++ const result = await recorder.startRtp(vod, video, audio);if (!result.ok) return res.status(409).json({ error: result.error });res.json({ videoPort: result.videoPort, audioPort: result.audioPort || null });} catch (err) {nnew file mode 100644@@@ -0,0 +1,62 @@++'use strict';++// A recording leaves the recorder when its ffmpeg ends. On 2026-09-28 the PostgreSQL switch made the++// registration async and the RTMP/RTP starters did not await it: the exit listener was attached to a++// Promise, so every recording stayed "in progress" after ffmpeg ended (15 of them, holding every deploy),++// and none was finalized. Here ffmpeg is a stand-in process: it exits normally, and it exits before the++// row write finishes; both times the recording leaves the active set.++const assert = require('assert');++const fs = require('fs');++const os = require('os');++const path = require('path');++const { EventEmitter } = require('events');++++const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-rec-'));++const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };++Object.assign(process.env, {++ DB_PATH: path.join(tmp, 'media.db'), VOD_PATH: dir('vods'), CLIPS_PATH: dir('clips'), FILES_PATH: dir('files'),++ THUMBNAILS_PATH: dir('thumbnails'), PASTES_PATH: dir('pastes'), OBJECTS_PATH: dir('objects'),++});++delete process.env.MEDIA_RTMP_PULL_ALLOW;++++// ffmpeg stand-in, installed before the recorder takes its reference to spawn.++const cp = require('child_process');++const spawned = [];++let exitAtOnce = false;++cp.spawn = () => {++ const p = new EventEmitter();++ p.stderr = new EventEmitter(); p.stdout = new EventEmitter(); p.killed = false;++ p.kill = () => { p.killed = true; setImmediate(() => p.emit('exit', 0, null)); return true; };++ if (exitAtOnce) setImmediate(() => p.emit('exit', 1, null));++ spawned.push(p);++ return p;++};++++(async () => {++ const recorder = require('../server/vod/recorder');++ const db = require('../server/db/database');++ await db.upsertApp({ app_id: 'live', api_key: 'live-key-rec-test' });++ const settled = async () => { for (let i = 0; i < 100 && recorder.activeCount(); i++) await new Promise((res) => setTimeout(res, 10)); };++ const newVod = async () => db.getVodById(Number((await db.createVod({ app_id: 'live', title: 'x' })).lastInsertRowid));++++ const a = await newVod();++ const r = await recorder.startRtmp(a, 'rtmp://127.0.0.1:1935/live/k');++ assert.strictEqual(r.ok, true);++ assert.strictEqual(recorder.activeCount(), 1);++ assert.strictEqual((await db.getVodById(a.id)).is_recording, 1, 'the row says so once start returns');++ assert.strictEqual(spawned.length, 1, 'the stand-in ran'); assert.strictEqual(recorder.stopRecording(a.id), true);++ await settled();++ assert.strictEqual(recorder.activeCount(), 0, 'ffmpeg ended: the recording is no longer in progress');++ console.log('✅ a stopped recording leaves the active set when ffmpeg exits');++++ exitAtOnce = true;++ const b = await newVod();++ assert.strictEqual((await recorder.startRtmp(b, 'rtmp://127.0.0.1:1935/live/k')).ok, true);++ await settled();++ assert.strictEqual(recorder.activeCount(), 0, 'an ffmpeg that dies at once is caught too');++ console.log('✅ an ffmpeg that exits while the row is being written is caught');++++ await db.close();++ fs.rmSync(tmp, { recursive: true, force: true });++ console.log('\n✅ recorder lifecycle tests passed');++ process.exit(0);++})().catch((err) => { console.error(err); process.exit(1); });@@@ -59,7 +59,7 @@ const express = require('express');const db = require('../server/db/database');await db.upsertApp({ app_id: 'live', api_key: 'live-key-rtmp-test' });const vodId = Number((await db.createVod({ app_id: 'live', title: 'x' })).lastInsertRowid);-- const r = recorder.startRtmp(await db.getVodById(vodId), 'rtmp://10.0.0.5:1935/live/k');++ const r = await recorder.startRtmp(await db.getVodById(vodId), 'rtmp://10.0.0.5:1935/live/k');assert.deepStrictEqual([r.ok, r.status], [false, 400]);assert.strictEqual(recorder.activeCount(), 0, 'no ffmpeg was started');assert.strictEqual((await db.getVodById(vodId)).is_recording, 0);
@@ -103,7 +103,7 @@ function createIngestHandler({ db, config, screenshotsDir, generateSlug, log = c
@@ -103,7 +103,7 @@ function createIngestHandler({ db, config, screenshotsDir, generateSlug, log = c
fs.mkdirSync(screenshotsDir, { recursive: true });
fs.mkdirSync(screenshotsDir, { recursive: true });
const file = path.join(screenshotsDir, `avatar-n${parseInt(user_id, 10) || 0}-${Date.now()}-${crypto.randomBytes(4).toString('hex')}.webp`);
const file = path.join(screenshotsDir, `avatar-n${parseInt(user_id, 10) || 0}-${Date.now()}-${crypto.randomBytes(4).toString('hex')}.webp`);
fs.writeFileSync(file, pic.buffer);
fs.writeFileSync(file, pic.buffer);
- const slug = generateSlug();
+ const slug = await generateSlug(); // async since the PostgreSQL switch (it checks the table)
const insert = async () => await db.run(`INSERT INTO pastes (app_id, slug, user_id, type, title, content, language, visibility, screenshot_path, metadata, ip_address)
const insert = async () => await db.run(`INSERT INTO pastes (app_id, slug, user_id, type, title, content, language, visibility, screenshot_path, metadata, ip_address)
VALUES ('network', ?, ?, 'screenshot', ?, '', 'text', 'unlisted', ?, ?, NULL) RETURNING id`,
VALUES ('network', ?, ?, 'screenshot', ?, '', 'text', 'unlisted', ?, ?, NULL) RETURNING id`,
[slug, parseInt(user_id, 10) || null, `Avatar${username ? ' of ' + String(username).slice(0, 40) : ''}`, file,
[slug, parseInt(user_id, 10) || null, `Avatar${username ? ' of ' + String(username).slice(0, 40) : ''}`, file,
@@ -333,7 +333,7 @@ class StreamRecorder {
@@ -333,7 +333,7 @@ class StreamRecorder {
* Exactly the predecessor's RTMP path — no re-encode, always real-time,
* Exactly the predecessor's RTMP path — no re-encode, always real-time,
* live-seekable, kill-tolerant, its own lossless master.
* live-seekable, kill-tolerant, its own lossless master.
*/
*/
- startRtmp(vod, rtmpUrl) {
+ async startRtmp(vod, rtmpUrl) {
const guard = this._guardCanRecord(vod);
const guard = this._guardCanRecord(vod);
if (!guard.ok) return guard;
if (!guard.ok) return guard;
// SSRF guard: ffmpeg connects only to an allow-listed RTMP server (config.rtmpPull.allow).
// SSRF guard: ffmpeg connects only to an allow-listed RTMP server (config.rtmpPull.allow).
@@ -368,8 +368,12 @@ class StreamRecorder {
@@ -368,8 +368,12 @@ class StreamRecorder {
return { ok: false, error: `ffmpeg spawn failed: ${err.message}` };
return { ok: false, error: `ffmpeg spawn failed: ${err.message}` };
}
}
- const recording = this._registerCommon(vod, proc, filePath, null, { protocol: 'rtmp' });
- this._wireProcess(recording, 'rtmp');
+ // Listeners first, then the row: _registerCommon puts the recording in the map before its first await,
+ // so ffmpeg's exit is caught even if it comes while the row is being written. (Unawaited, this left every
+ // recording "in progress" forever after ffmpeg ended, and nothing finalized it.)
+ const persisted = this._registerCommon(vod, proc, filePath, null, { protocol: 'rtmp' });
+ this._wireProcess(this.activeRecordings.get(vod.id), 'rtmp');
+ await persisted;
console.log(`[VOD] Recording started: vod ${vod.id} → ${filename} (rtmp pull)`);
console.log(`[VOD] Recording started: vod ${vod.id} → ${filename} (rtmp pull)`);
return { ok: true, filePath };
return { ok: true, filePath };
@@ -384,7 +388,7 @@ class StreamRecorder {
@@ -384,7 +388,7 @@ class StreamRecorder {
* @param {object} [audio] same shape (codec e.g. 'opus')
* @param {object} [audio] same shape (codec e.g. 'opus')
* @returns {ok, videoPort, audioPort} — caller points PlainRtpTransports at 127.0.0.1
* @returns {ok, videoPort, audioPort} — caller points PlainRtpTransports at 127.0.0.1
*/
*/
- startRtp(vod, video, audio) {
+ async startRtp(vod, video, audio) {
const guard = this._guardCanRecord(vod);
const guard = this._guardCanRecord(vod);
if (!guard.ok) return guard;
if (!guard.ok) return guard;
if (!video || video.payloadType == null || !video.codec) {
if (!video || video.payloadType == null || !video.codec) {
@@ -514,13 +518,14 @@ class StreamRecorder {
@@ -514,13 +518,14 @@ class StreamRecorder {
return { ok: false, error: `ffmpeg spawn failed: ${err.message}` };
return { ok: false, error: `ffmpeg spawn failed: ${err.message}` };
}
}
- const recording = this._registerCommon(vod, proc, filePath, masterPath, {
+ const persisted = this._registerCommon(vod, proc, filePath, masterPath, {
protocol: 'rtp',
protocol: 'rtp',
sdpPath,
sdpPath,
rtpVideoPort: videoPort,
rtpVideoPort: videoPort,
rtpAudioPort: audioPort,
rtpAudioPort: audioPort,
});
});
- this._wireProcess(recording, 'rtp');
+ this._wireProcess(this.activeRecordings.get(vod.id), 'rtp'); // before the row write (see startRtmp)
+ await persisted;
const recMode = passthrough ? `passthrough ${isH264 ? 'H.264→mp4' : 'VP8/9→webm'} copy` : 'libvpx re-encode';
const recMode = passthrough ? `passthrough ${isH264 ? 'H.264→mp4' : 'VP8/9→webm'} copy` : 'libvpx re-encode';
console.log(`[VOD] RTP recording started: vod ${vod.id} → ${filename} [${recMode}] (video:${videoPort}${audioPort ? ` audio:${audioPort}` : ''})`);
console.log(`[VOD] RTP recording started: vod ${vod.id} → ${filename} [${recMode}] (video:${videoPort}${audioPort ? ` audio:${audioPort}` : ''})`);
@@ -122,7 +122,7 @@ router.post('/:id/ingest/rtmp', tenantAuth(), async (req, res) => {
@@ -122,7 +122,7 @@ router.post('/:id/ingest/rtmp', tenantAuth(), async (req, res) => {
try {
try {
const vod = await _getVodForWrite(req, res);
const vod = await _getVodForWrite(req, res);
if (!vod) return;
if (!vod) return;
- const result = recorder.startRtmp(vod, req.body?.rtmp_url);
+ const result = await recorder.startRtmp(vod, req.body?.rtmp_url);
if (!result.ok) return res.status(result.status || 409).json({ error: result.error });
if (!result.ok) return res.status(result.status || 409).json({ error: result.error });
res.status(202).json({ id: vod.id, status: 'recording' });
res.status(202).json({ id: vod.id, status: 'recording' });
} catch (err) {
} catch (err) {
@@ -137,7 +137,7 @@ router.post('/:id/ingest/rtp/start', tenantAuth(), async (req, res) => {
@@ -137,7 +137,7 @@ router.post('/:id/ingest/rtp/start', tenantAuth(), async (req, res) => {
const vod = await _getVodForWrite(req, res);
const vod = await _getVodForWrite(req, res);
if (!vod) return;
if (!vod) return;
const { video, audio } = req.body || {};
const { video, audio } = req.body || {};
- const result = recorder.startRtp(vod, video, audio);
+ const result = await recorder.startRtp(vod, video, audio);
if (!result.ok) return res.status(409).json({ error: result.error });
if (!result.ok) return res.status(409).json({ error: result.error });
res.json({ videoPort: result.videoPort, audioPort: result.audioPort || null });
res.json({ videoPort: result.videoPort, audioPort: result.audioPort || null });
} catch (err) {
} catch (err) {
new file mode 100644
new file mode 100644
@@ -0,0 +1,62 @@
@@ -0,0 +1,62 @@
+'use strict';
+// A recording leaves the recorder when its ffmpeg ends. On 2026-09-28 the PostgreSQL switch made the
+// registration async and the RTMP/RTP starters did not await it: the exit listener was attached to a
+// Promise, so every recording stayed "in progress" after ffmpeg ended (15 of them, holding every deploy),
+// and none was finalized. Here ffmpeg is a stand-in process: it exits normally, and it exits before the
+// row write finishes; both times the recording leaves the active set.
+const assert = require('assert');
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+const { EventEmitter } = require('events');
+
+const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-rec-'));
+const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };
+Object.assign(process.env, {
+ DB_PATH: path.join(tmp, 'media.db'), VOD_PATH: dir('vods'), CLIPS_PATH: dir('clips'), FILES_PATH: dir('files'),
+ THUMBNAILS_PATH: dir('thumbnails'), PASTES_PATH: dir('pastes'), OBJECTS_PATH: dir('objects'),
+});
+delete process.env.MEDIA_RTMP_PULL_ALLOW;
+
+// ffmpeg stand-in, installed before the recorder takes its reference to spawn.
+const cp = require('child_process');
+const spawned = [];
+let exitAtOnce = false;
+cp.spawn = () => {
+ const p = new EventEmitter();
+ p.stderr = new EventEmitter(); p.stdout = new EventEmitter(); p.killed = false;
+ p.kill = () => { p.killed = true; setImmediate(() => p.emit('exit', 0, null)); return true; };
+ if (exitAtOnce) setImmediate(() => p.emit('exit', 1, null));
+ spawned.push(p);
+ return p;
+};
+
+(async () => {
+ const recorder = require('../server/vod/recorder');
+ const db = require('../server/db/database');
+ await db.upsertApp({ app_id: 'live', api_key: 'live-key-rec-test' });
+ const settled = async () => { for (let i = 0; i < 100 && recorder.activeCount(); i++) await new Promise((res) => setTimeout(res, 10)); };
+ const newVod = async () => db.getVodById(Number((await db.createVod({ app_id: 'live', title: 'x' })).lastInsertRowid));
+
+ const a = await newVod();
+ const r = await recorder.startRtmp(a, 'rtmp://127.0.0.1:1935/live/k');
+ assert.strictEqual(r.ok, true);
+ assert.strictEqual(recorder.activeCount(), 1);
+ assert.strictEqual((await db.getVodById(a.id)).is_recording, 1, 'the row says so once start returns');
+ assert.strictEqual(spawned.length, 1, 'the stand-in ran'); assert.strictEqual(recorder.stopRecording(a.id), true);
+ await settled();
+ assert.strictEqual(recorder.activeCount(), 0, 'ffmpeg ended: the recording is no longer in progress');
+ console.log('✅ a stopped recording leaves the active set when ffmpeg exits');
+
+ exitAtOnce = true;
+ const b = await newVod();
+ assert.strictEqual((await recorder.startRtmp(b, 'rtmp://127.0.0.1:1935/live/k')).ok, true);
+ await settled();
+ assert.strictEqual(recorder.activeCount(), 0, 'an ffmpeg that dies at once is caught too');
+ console.log('✅ an ffmpeg that exits while the row is being written is caught');
+
+ await db.close();
+ fs.rmSync(tmp, { recursive: true, force: true });
+ console.log('\n✅ recorder lifecycle tests passed');
+ process.exit(0);
+})().catch((err) => { console.error(err); process.exit(1); });
@@ -59,7 +59,7 @@ const express = require('express');
@@ -59,7 +59,7 @@ const express = require('express');
const db = require('../server/db/database');
const db = require('../server/db/database');
await db.upsertApp({ app_id: 'live', api_key: 'live-key-rtmp-test' });
await db.upsertApp({ app_id: 'live', api_key: 'live-key-rtmp-test' });
const vodId = Number((await db.createVod({ app_id: 'live', title: 'x' })).lastInsertRowid);
const vodId = Number((await db.createVod({ app_id: 'live', title: 'x' })).lastInsertRowid);
- const r = recorder.startRtmp(await db.getVodById(vodId), 'rtmp://10.0.0.5:1935/live/k');
+ const r = await recorder.startRtmp(await db.getVodById(vodId), 'rtmp://10.0.0.5:1935/live/k');
assert.deepStrictEqual([r.ok, r.status], [false, 400]);
assert.deepStrictEqual([r.ok, r.status], [false, 400]);
assert.strictEqual(recorder.activeCount(), 0, 'no ffmpeg was started');
assert.strictEqual(recorder.activeCount(), 0, 'no ffmpeg was started');
assert.strictEqual((await db.getVodById(vodId)).is_recording, 0);
assert.strictEqual((await db.getVodById(vodId)).is_recording, 0);