openvibers/openvibe.live Commits
Recordings: retry a disk-refused start while the stream is live; relay Media storage alerts
Changes
@@@ -85,3 +85,8 @@ MAX_VOD_SIZE_MB=2048# Word FilterUNSAFE_WORDS_PATH=./server/chat/unsafe-words.txt++++# Ops alerts (optional) — Discord-compatible webhook that receives storage alerts relayed++# from OpenVibe.Media (drain stalled / disk critical / recovered). Can also be set from the++# admin settings as `ops_alert_webhook_url`.++OPS_ALERT_WEBHOOK_URL=@@@ -32,6 +32,40 @@ function verifySignature(req) {}}++/**++ * Storage alerts from Media (`storage.alert` / `storage.recovered`). Always logged at++ * error/warn level; additionally posted to an ops webhook when one is configured —++ * OPS_ALERT_WEBHOOK_URL in the environment, or the `ops_alert_webhook_url` admin++ * setting (a Discord-compatible webhook: it receives `{ content }`). This is++ * deliberately separate from the public go-live Discord webhook.++ */++function relayStorageEvent(event, data = {}) {++ const recovered = event === 'storage.recovered';++ const line = `[MediaWebhook] ${recovered ? 'STORAGE RECOVERED' : 'STORAGE ALERT'} (${data.kind || 'unknown'}): `++ + `disk ${data.disk_pct ?? '?'}%, ${data.free_gb ?? '?'} GB free`++ + (data.hint ? ` — ${data.hint}` : '');++ (recovered ? console.warn : console.error)(line);++ if (!recovered && Array.isArray(data.errors) && data.errors.length) {++ console.error('[MediaWebhook] first errors:', JSON.stringify(data.errors));++ }++++ let url = process.env.OPS_ALERT_WEBHOOK_URL || '';++ if (!url) { try { url = db.getSetting('ops_alert_webhook_url') || ''; } catch { url = ''; } }++ if (!/^https:\/\/[^\s]+$/i.test(url)) return;++++ const content = (recovered ? '✅ **OpenVibe.Media storage recovered**' : '🚨 **OpenVibe.Media storage alert**')++ + `\n${data.kind || 'unknown'} — disk ${data.disk_pct ?? '?'}%, ${data.free_gb ?? '?'} GB free`++ + (data.stalled_passes ? `, ${data.stalled_passes} stalled sweep pass(es)` : '')++ + (data.hint ? `\n${data.hint}` : '')++ + (Array.isArray(data.errors) && data.errors.length ? `\n\`${JSON.stringify(data.errors).slice(0, 300)}\`` : '');++ fetch(url, {++ method: 'POST',++ headers: { 'Content-Type': 'application/json' },++ body: JSON.stringify({ content }),++ signal: AbortSignal.timeout(10000),++ }).catch(err => console.warn('[MediaWebhook] ops alert webhook failed:', err.message));++}++function handler(req, res) {if (!verifySignature(req)) return res.status(401).json({ error: 'Invalid signature' });const { event, data } = req.body || {};@@@ -91,6 +125,13 @@ function handler(req, res) {if (data?.id) { try { db.setClipTranscriptStatus(data.id, 'failed', data?.error || 'media reported failure'); } catch { /* */ } }console.warn(`[MediaWebhook] Clip ${data?.id} failed:`, data?.error || '(no detail)');break;++ case 'storage.alert':++ case 'storage.recovered':++ // Media's VOD volume needs a human (drain stalled / disk critical) or is++ // fine again. Recordings are silently refused while this is unresolved,++ // which is why it is loud here and forwarded to the ops channel.++ relayStorageEvent(event, data);++ break;default:console.log(`[MediaWebhook] Ignoring event: ${event}`);}@@@ -36,6 +36,10 @@ class StreamRecorder {this.activeRecordings = new Map();this._finalizing = new Set();this._healAttempts = new Map();++ // streamId → timer for a recording start that Media refused for lack of disk;++ // retried while the stream stays live so the drain freeing space still yields++ // a (partial) recording instead of none at all.++ this._diskRetryTimers = new Map();}_title(stream, opts = {}) {@@@ -95,9 +99,43 @@ class StreamRecorder {// ghost with no file that shows up as a 0:00 VOD and — worse — clogs Media's// offload sweep, which is exactly what needs to run when the disk is full.this._abort(streamId, rec).catch(() => {});++ this._maybeRetryAfterDiskRefusal(streamId, protocol, endpoint, opts, err);});}++ static get DISK_RETRY_DELAY_MS() { return 5 * 60 * 1000; }++ static get DISK_RETRY_MAX() { return 12; } // ≈ 1 h of a still-live stream++++ /**++ * Media refuses to start a recording while its VOD volume is critically low. Its++ * offload sweep frees space within minutes, so keep trying while the stream is live++ * — a recording that starts 10 minutes in beats no recording at all.++ */++ _maybeRetryAfterDiskRefusal(streamId, protocol, endpoint, opts, err) {++ if (!/disk/i.test(String(err?.message || ''))) return;++ const attempt = (opts._diskRetries || 0) + 1;++ if (attempt > StreamRecorder.DISK_RETRY_MAX) {++ console.error(`[VOD] Giving up on recording stream ${streamId} — Media still refusing for disk space after ${attempt - 1} retries`);++ return;++ }++ this._clearDiskRetry(streamId);++ const timer = setTimeout(() => {++ this._diskRetryTimers.delete(streamId);++ const live = db.getStreamById(streamId);++ if (!live || !live.is_live) return;++ console.log(`[VOD] Retrying recording start for stream ${streamId} (disk-space retry ${attempt}/${StreamRecorder.DISK_RETRY_MAX})`);++ this.startRecording(streamId, protocol, endpoint, { ...opts, _diskRetries: attempt });++ }, StreamRecorder.DISK_RETRY_DELAY_MS);++ if (timer.unref) timer.unref();++ this._diskRetryTimers.set(streamId, timer);++ console.warn(`[VOD] Will retry recording for stream ${streamId} in ${StreamRecorder.DISK_RETRY_DELAY_MS / 60000} min (disk-space refusal, attempt ${attempt}/${StreamRecorder.DISK_RETRY_MAX})`);++ }++++ _clearDiskRetry(streamId) {++ const timer = this._diskRetryTimers.get(streamId);++ if (timer) { clearTimeout(timer); this._diskRetryTimers.delete(streamId); }++ }++async _createVod(streamId, stream, rec, opts) {const { id } = await media.createVod({title: this._title(stream, opts),@@@ -249,6 +287,7 @@ class StreamRecorder {* Stop (and finalize) the recording for a stream. Safe to call repeatedly.*/stopRecording(streamId) {++ this._clearDiskRetry(streamId); // stream ending cancels any pending disk-space retryconst rec = this.activeRecordings.get(streamId);if (!rec) return;rec._cancel = true;nnew file mode 100644@@@ -0,0 +1,71 @@++'use strict';++++// When Media refuses a recording because its VOD volume is critically low, the++// recorder must (a) drop the empty VOD shell it created and (b) keep retrying while++// the stream is live, because Media's offload sweep frees space within minutes.++// Anything that is not a disk refusal must not be retried.++++const assert = require('assert');++const fs = require('fs');++const os = require('os');++const path = require('path');++++process.env.DB_PATH = path.join(fs.mkdtempSync(path.join(os.tmpdir(), 'ov-rec-')), 'live.db');++const db = require('../server/db/database');++const media = require('../server/media-client');++const recorder = require('../server/streaming/recorder');++++// Stubs: the recorder only needs to know the stream exists and whether it is live.++let live = true;++db.getStreamById = (id) => ({ id, user_id: 1, title: 'T', is_live: live ? 1 : 0, managed_stream_key: 'k' });++db.resolveStreamVodVisibility = () => 'public';++const deleted = [];++media.deleteVod = async (id) => { deleted.push(id); };++++// _abort is what drops the shell; make sure the failure path reaches it.++const abortCalls = [];++const origAbort = recorder._abort.bind(recorder);++recorder._abort = async (streamId, rec) => { abortCalls.push({ streamId, vodId: rec.vodId }); return origAbort(streamId, rec); };++++// A disk refusal schedules a retry and clears the active slot.++recorder._maybeRetryAfterDiskRefusal(42, 'rtmp', { streamKey: 'k' }, {}, new Error('Disk critically low — recording refused'));++assert.ok(recorder._diskRetryTimers.has(42), 'disk refusal schedules a retry timer');++console.log('✅ disk-space refusal schedules a retry');++++// The stream ending cancels the retry.++recorder.stopRecording(42);++assert.ok(!recorder._diskRetryTimers.has(42), 'stopRecording cancels the pending retry');++console.log('✅ stream end cancels the pending retry');++++// Other failures are not retried.++recorder._maybeRetryAfterDiskRefusal(43, 'rtmp', { streamKey: 'k' }, {}, new Error('Media unreachable (POST /vods): fetch failed'));++assert.ok(!recorder._diskRetryTimers.has(43), 'non-disk failures are not retried');++console.log('✅ non-disk failures are not retried');++++// Retries are capped.++recorder._maybeRetryAfterDiskRefusal(44, 'rtmp', { streamKey: 'k' }, { _diskRetries: 12 }, new Error('Disk critically low — recording refused'));++assert.ok(!recorder._diskRetryTimers.has(44), 'gives up after the retry cap');++console.log('✅ retry cap honoured');++++// Rescheduling replaces rather than stacks timers.++recorder._maybeRetryAfterDiskRefusal(45, 'rtmp', { streamKey: 'k' }, {}, new Error('disk full'));++const first = recorder._diskRetryTimers.get(45);++recorder._maybeRetryAfterDiskRefusal(45, 'rtmp', { streamKey: 'k' }, { _diskRetries: 1 }, new Error('disk full'));++assert.notStrictEqual(recorder._diskRetryTimers.get(45), first, 'a new retry replaces the old timer');++recorder._clearDiskRetry(45);++console.log('✅ retries replace, never stack');++++// The start-failure path drops the VOD shell in Media (the ghost-row fix).++media.createVod = async () => ({ id: 777 });++media.ingestRtmp = async () => { throw new Error('Disk critically low — recording refused'); };++recorder.startRecording(46, 'rtmp', { streamKey: 'k' }, { mode: 'vod' });++setTimeout(() => {++ assert.deepStrictEqual(abortCalls.map(a => a.vodId), [777], '_abort ran with the created VOD id');++ assert.deepStrictEqual(deleted, [777], 'the empty VOD shell was deleted in Media');++ assert.ok(!recorder.isRecording(46), 'no active recording is left behind');++ assert.ok(recorder._diskRetryTimers.has(46), 'and a retry is scheduled because the refusal was disk-related');++ recorder._clearDiskRetry(46);++ console.log('✅ start failure deletes the Media VOD shell and schedules a retry');++ console.log('\n✅ All recorder disk-retry tests passed');++ process.exit(0);++}, 200);
@@ -85,3 +85,8 @@ MAX_VOD_SIZE_MB=2048
@@ -85,3 +85,8 @@ MAX_VOD_SIZE_MB=2048
# Word Filter
# Word Filter
UNSAFE_WORDS_PATH=./server/chat/unsafe-words.txt
UNSAFE_WORDS_PATH=./server/chat/unsafe-words.txt
+
+# Ops alerts (optional) — Discord-compatible webhook that receives storage alerts relayed
+# from OpenVibe.Media (drain stalled / disk critical / recovered). Can also be set from the
+# admin settings as `ops_alert_webhook_url`.
+OPS_ALERT_WEBHOOK_URL=
@@ -32,6 +32,40 @@ function verifySignature(req) {
@@ -32,6 +32,40 @@ function verifySignature(req) {
}
}
}
}
+/**
+ * Storage alerts from Media (`storage.alert` / `storage.recovered`). Always logged at
+ * error/warn level; additionally posted to an ops webhook when one is configured —
+ * OPS_ALERT_WEBHOOK_URL in the environment, or the `ops_alert_webhook_url` admin
+ * setting (a Discord-compatible webhook: it receives `{ content }`). This is
+ * deliberately separate from the public go-live Discord webhook.
+ */
+function relayStorageEvent(event, data = {}) {
+ const recovered = event === 'storage.recovered';
+ const line = `[MediaWebhook] ${recovered ? 'STORAGE RECOVERED' : 'STORAGE ALERT'} (${data.kind || 'unknown'}): `
+ + `disk ${data.disk_pct ?? '?'}%, ${data.free_gb ?? '?'} GB free`
+ + (data.hint ? ` — ${data.hint}` : '');
+ (recovered ? console.warn : console.error)(line);
+ if (!recovered && Array.isArray(data.errors) && data.errors.length) {
+ console.error('[MediaWebhook] first errors:', JSON.stringify(data.errors));
+ }
+
+ let url = process.env.OPS_ALERT_WEBHOOK_URL || '';
+ if (!url) { try { url = db.getSetting('ops_alert_webhook_url') || ''; } catch { url = ''; } }
+ if (!/^https:\/\/[^\s]+$/i.test(url)) return;
+
+ const content = (recovered ? '✅ **OpenVibe.Media storage recovered**' : '🚨 **OpenVibe.Media storage alert**')
+ + `\n${data.kind || 'unknown'} — disk ${data.disk_pct ?? '?'}%, ${data.free_gb ?? '?'} GB free`
+ + (data.stalled_passes ? `, ${data.stalled_passes} stalled sweep pass(es)` : '')
+ + (data.hint ? `\n${data.hint}` : '')
+ + (Array.isArray(data.errors) && data.errors.length ? `\n\`${JSON.stringify(data.errors).slice(0, 300)}\`` : '');
+ fetch(url, {
+ method: 'POST',
+ headers: { 'Content-Type': 'application/json' },
+ body: JSON.stringify({ content }),
+ signal: AbortSignal.timeout(10000),
+ }).catch(err => console.warn('[MediaWebhook] ops alert webhook failed:', err.message));
+}
+
function handler(req, res) {
function handler(req, res) {
if (!verifySignature(req)) return res.status(401).json({ error: 'Invalid signature' });
if (!verifySignature(req)) return res.status(401).json({ error: 'Invalid signature' });
const { event, data } = req.body || {};
const { event, data } = req.body || {};
@@ -91,6 +125,13 @@ function handler(req, res) {
@@ -91,6 +125,13 @@ function handler(req, res) {
if (data?.id) { try { db.setClipTranscriptStatus(data.id, 'failed', data?.error || 'media reported failure'); } catch { /* */ } }
if (data?.id) { try { db.setClipTranscriptStatus(data.id, 'failed', data?.error || 'media reported failure'); } catch { /* */ } }
console.warn(`[MediaWebhook] Clip ${data?.id} failed:`, data?.error || '(no detail)');
console.warn(`[MediaWebhook] Clip ${data?.id} failed:`, data?.error || '(no detail)');
break;
break;
+ case 'storage.alert':
+ case 'storage.recovered':
+ // Media's VOD volume needs a human (drain stalled / disk critical) or is
+ // fine again. Recordings are silently refused while this is unresolved,
+ // which is why it is loud here and forwarded to the ops channel.
+ relayStorageEvent(event, data);
+ break;
default:
default:
console.log(`[MediaWebhook] Ignoring event: ${event}`);
console.log(`[MediaWebhook] Ignoring event: ${event}`);
}
}
@@ -36,6 +36,10 @@ class StreamRecorder {
@@ -36,6 +36,10 @@ class StreamRecorder {
this.activeRecordings = new Map();
this.activeRecordings = new Map();
this._finalizing = new Set();
this._finalizing = new Set();
this._healAttempts = new Map();
this._healAttempts = new Map();
+ // streamId → timer for a recording start that Media refused for lack of disk;
+ // retried while the stream stays live so the drain freeing space still yields
+ // a (partial) recording instead of none at all.
+ this._diskRetryTimers = new Map();
}
}
_title(stream, opts = {}) {
_title(stream, opts = {}) {
@@ -95,9 +99,43 @@ class StreamRecorder {
@@ -95,9 +99,43 @@ class StreamRecorder {
// ghost with no file that shows up as a 0:00 VOD and — worse — clogs Media's
// ghost with no file that shows up as a 0:00 VOD and — worse — clogs Media's
// offload sweep, which is exactly what needs to run when the disk is full.
// offload sweep, which is exactly what needs to run when the disk is full.
this._abort(streamId, rec).catch(() => {});
this._abort(streamId, rec).catch(() => {});
+ this._maybeRetryAfterDiskRefusal(streamId, protocol, endpoint, opts, err);
});
});
}
}
+ static get DISK_RETRY_DELAY_MS() { return 5 * 60 * 1000; }
+ static get DISK_RETRY_MAX() { return 12; } // ≈ 1 h of a still-live stream
+
+ /**
+ * Media refuses to start a recording while its VOD volume is critically low. Its
+ * offload sweep frees space within minutes, so keep trying while the stream is live
+ * — a recording that starts 10 minutes in beats no recording at all.
+ */
+ _maybeRetryAfterDiskRefusal(streamId, protocol, endpoint, opts, err) {
+ if (!/disk/i.test(String(err?.message || ''))) return;
+ const attempt = (opts._diskRetries || 0) + 1;
+ if (attempt > StreamRecorder.DISK_RETRY_MAX) {
+ console.error(`[VOD] Giving up on recording stream ${streamId} — Media still refusing for disk space after ${attempt - 1} retries`);
+ return;
+ }
+ this._clearDiskRetry(streamId);
+ const timer = setTimeout(() => {
+ this._diskRetryTimers.delete(streamId);
+ const live = db.getStreamById(streamId);
+ if (!live || !live.is_live) return;
+ console.log(`[VOD] Retrying recording start for stream ${streamId} (disk-space retry ${attempt}/${StreamRecorder.DISK_RETRY_MAX})`);
+ this.startRecording(streamId, protocol, endpoint, { ...opts, _diskRetries: attempt });
+ }, StreamRecorder.DISK_RETRY_DELAY_MS);
+ if (timer.unref) timer.unref();
+ this._diskRetryTimers.set(streamId, timer);
+ console.warn(`[VOD] Will retry recording for stream ${streamId} in ${StreamRecorder.DISK_RETRY_DELAY_MS / 60000} min (disk-space refusal, attempt ${attempt}/${StreamRecorder.DISK_RETRY_MAX})`);
+ }
+
+ _clearDiskRetry(streamId) {
+ const timer = this._diskRetryTimers.get(streamId);
+ if (timer) { clearTimeout(timer); this._diskRetryTimers.delete(streamId); }
+ }
+
async _createVod(streamId, stream, rec, opts) {
async _createVod(streamId, stream, rec, opts) {
const { id } = await media.createVod({
const { id } = await media.createVod({
title: this._title(stream, opts),
title: this._title(stream, opts),
@@ -249,6 +287,7 @@ class StreamRecorder {
@@ -249,6 +287,7 @@ class StreamRecorder {
* Stop (and finalize) the recording for a stream. Safe to call repeatedly.
* Stop (and finalize) the recording for a stream. Safe to call repeatedly.
*/
*/
stopRecording(streamId) {
stopRecording(streamId) {
+ this._clearDiskRetry(streamId); // stream ending cancels any pending disk-space retry
const rec = this.activeRecordings.get(streamId);
const rec = this.activeRecordings.get(streamId);
if (!rec) return;
if (!rec) return;
rec._cancel = true;
rec._cancel = true;
new file mode 100644
new file mode 100644
@@ -0,0 +1,71 @@
@@ -0,0 +1,71 @@
+'use strict';
+
+// When Media refuses a recording because its VOD volume is critically low, the
+// recorder must (a) drop the empty VOD shell it created and (b) keep retrying while
+// the stream is live, because Media's offload sweep frees space within minutes.
+// Anything that is not a disk refusal must not be retried.
+
+const assert = require('assert');
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+
+process.env.DB_PATH = path.join(fs.mkdtempSync(path.join(os.tmpdir(), 'ov-rec-')), 'live.db');
+const db = require('../server/db/database');
+const media = require('../server/media-client');
+const recorder = require('../server/streaming/recorder');
+
+// Stubs: the recorder only needs to know the stream exists and whether it is live.
+let live = true;
+db.getStreamById = (id) => ({ id, user_id: 1, title: 'T', is_live: live ? 1 : 0, managed_stream_key: 'k' });
+db.resolveStreamVodVisibility = () => 'public';
+const deleted = [];
+media.deleteVod = async (id) => { deleted.push(id); };
+
+// _abort is what drops the shell; make sure the failure path reaches it.
+const abortCalls = [];
+const origAbort = recorder._abort.bind(recorder);
+recorder._abort = async (streamId, rec) => { abortCalls.push({ streamId, vodId: rec.vodId }); return origAbort(streamId, rec); };
+
+// A disk refusal schedules a retry and clears the active slot.
+recorder._maybeRetryAfterDiskRefusal(42, 'rtmp', { streamKey: 'k' }, {}, new Error('Disk critically low — recording refused'));
+assert.ok(recorder._diskRetryTimers.has(42), 'disk refusal schedules a retry timer');
+console.log('✅ disk-space refusal schedules a retry');
+
+// The stream ending cancels the retry.
+recorder.stopRecording(42);
+assert.ok(!recorder._diskRetryTimers.has(42), 'stopRecording cancels the pending retry');
+console.log('✅ stream end cancels the pending retry');
+
+// Other failures are not retried.
+recorder._maybeRetryAfterDiskRefusal(43, 'rtmp', { streamKey: 'k' }, {}, new Error('Media unreachable (POST /vods): fetch failed'));
+assert.ok(!recorder._diskRetryTimers.has(43), 'non-disk failures are not retried');
+console.log('✅ non-disk failures are not retried');
+
+// Retries are capped.
+recorder._maybeRetryAfterDiskRefusal(44, 'rtmp', { streamKey: 'k' }, { _diskRetries: 12 }, new Error('Disk critically low — recording refused'));
+assert.ok(!recorder._diskRetryTimers.has(44), 'gives up after the retry cap');
+console.log('✅ retry cap honoured');
+
+// Rescheduling replaces rather than stacks timers.
+recorder._maybeRetryAfterDiskRefusal(45, 'rtmp', { streamKey: 'k' }, {}, new Error('disk full'));
+const first = recorder._diskRetryTimers.get(45);
+recorder._maybeRetryAfterDiskRefusal(45, 'rtmp', { streamKey: 'k' }, { _diskRetries: 1 }, new Error('disk full'));
+assert.notStrictEqual(recorder._diskRetryTimers.get(45), first, 'a new retry replaces the old timer');
+recorder._clearDiskRetry(45);
+console.log('✅ retries replace, never stack');
+
+// The start-failure path drops the VOD shell in Media (the ghost-row fix).
+media.createVod = async () => ({ id: 777 });
+media.ingestRtmp = async () => { throw new Error('Disk critically low — recording refused'); };
+recorder.startRecording(46, 'rtmp', { streamKey: 'k' }, { mode: 'vod' });
+setTimeout(() => {
+ assert.deepStrictEqual(abortCalls.map(a => a.vodId), [777], '_abort ran with the created VOD id');
+ assert.deepStrictEqual(deleted, [777], 'the empty VOD shell was deleted in Media');
+ assert.ok(!recorder.isRecording(46), 'no active recording is left behind');
+ assert.ok(recorder._diskRetryTimers.has(46), 'and a retry is scheduled because the refusal was disk-related');
+ recorder._clearDiskRetry(46);
+ console.log('✅ start failure deletes the Media VOD shell and schedules a retry');
+ console.log('\n✅ All recorder disk-retry tests passed');
+ process.exit(0);
+}, 200);