openvibers/openvibe.live Commits

Browser streams are recorded with their sound: the recorder waits for the microphone, and rolls to a part with it when it comes later (#373)

OP OpenVibers committed 18 days ago · d471196b09 · 3 files, +159 −19

VOD 5986 (goosely, 2026-10-25) has no audio track: the recorder found the video producer and asked for the audio one in
the same instant, a moment before the browser produced it (both in the same second), and recorded seven minutes of
video only. The restreams already restart to pick up late audio; the recorder never did.

- _startRtpRecording waits up to 5 s (VOD_AUDIO_WAIT_MS) for the audio producer after the video one; a stream with no
  microphone then records video only.
- The producer watch also follows audio: a microphone that arrives while the recording has no live one (none at the
  start, or the recorded one stopped) rolls the recording to the next part, which has the sound. A second microphone
  beside a live one is left alone. The roll is shared with the video reconnect (_rollRecording).
- docs/vods-and-clips.md: the protocol table says how recording works now, and the audio wait.

Checked: test/recorder-audio.test.js (5986's race, no microphone, a late one, a second one, a replaced one) and the
12 recorder tests the change reaches.

Changes

ddiff --git a/docs/vods-and-clips.md b/docs/vods-and-clips.md
iindex 7cd16756..72a46735 100644
---- a/docs/vods-and-clips.md
++++ b/docs/vods-and-clips.md
@@@ -29,9 +29,14 @@ OpenVibe.Live automatically records your streams as VODs (Video on Demand).
| Protocol | Recording Method | Format |
|----------|-----------------|--------|
--| **WebRTC (Browser)** | Browser-side MediaRecorder, chunks uploaded to server | WebM (VP8/VP9 + Opus) |
--| **RTMP** | Server-side FFmpeg capture | WebM (VP8 + Vorbis) |
--| **JSMPEG** | Server-side FFmpeg via WebSocket relay | WebM (VP8 + Vorbis) |
++| **WebRTC (Browser, WHIP)** | Server-side: the SFU forwards the broadcaster's video and audio as RTP to OpenVibe.Media, which copies them | WebM (VP8 + Opus), MP4 (H.264 + AAC) from phones |
++| **RTMP** | Server-side: OpenVibe.Media pulls the local RTMP endpoint and copies it | MP4 (H.264 + AAC) |
++| **JSMPEG** | Not recorded (Media has no MPEG-TS ingest) | — |
++
++A browser produces its audio a moment after its video. The recorder waits up to 5 s for it (`VOD_AUDIO_WAIT_MS`):
++VOD 5986 (2026-10-25) was recorded silent because the recorder asked for the audio in the same instant the video
++arrived, just before the microphone did. A stream with no microphone records video only after the wait. A microphone
++that comes later, or a new one after the recorded one stopped, rolls the recording to a new part that has the sound.
### Recording Lifecycle
ddiff --git a/server/streaming/recorder.js b/server/streaming/recorder.js
iindex 394aaf93..5cdcd8c7 100644
---- a/server/streaming/recorder.js
++++ b/server/streaming/recorder.js
@@@ -61,40 +61,57 @@ class StreamRecorder {
* Media keep "recording" nothing and the stream ends as a zero-byte VOD. Watch for a video
* producer arriving on a DIFFERENT transport than the one we recorded from, and roll the
* recording to a new part on the new producer. A second video source on the same transport
-- * (screen share next to the camera) is not a reconnect and is left alone.
++ * (screen share next to the camera) is not a reconnect and is left alone. Likewise a microphone
++ * that arrives after the recording started without sound rolls it to a part with sound.
*/
_installProducerWatch() {
if (this._producerWatch) return;
let webrtcSFU;
try { webrtcSFU = require('./webrtc-sfu'); } catch (e) { console.warn('[VOD] producer watch not installed:', e.message); return; }
this._producerWatch = ({ roomId, producerId, kind }) => {
-- if (kind !== 'video') return;
++ if (kind !== 'video' && kind !== 'audio') return;
const m = /^stream-(\d+)$/.exec(String(roomId || ''));
if (!m) return;
const streamId = Number(m[1]);
const rec = this.activeRecordings.get(streamId);
-- if (!rec || rec.type !== 'rtp' || !rec.videoProducerId || rec.videoProducerId === producerId || rec._rolling) return;
++ if (!rec || rec.type !== 'rtp' || !rec.videoProducerId || rec._rolling) return;
const room = webrtcSFU.rooms?.get(roomId);
++ const alive = (id) => { const e = id && room?.producers?.get(id); return !!(e && e.producer && !e.producer.closed); };
++ if (kind === 'audio') {
++ // A microphone that arrives after the recording settled without one (or after its own died): the
++ // recording so far is silent, so it rolls to a part that has the sound. A second microphone beside a
++ // live one is not followed.
++ if (!rec.audioSettled || rec.audioProducerId === producerId || alive(rec.audioProducerId)) return;
++ this._rollRecording(streamId, rec, 'a microphone arrived while the recording had no live one');
++ return;
++ }
++ if (rec.videoProducerId === producerId) return;
const oldEntry = room?.producers?.get(rec.videoProducerId);
const newEntry = room?.producers?.get(producerId);
const oldAlive = !!(oldEntry && oldEntry.producer && !oldEntry.producer.closed);
const sameTransport = !!(oldEntry && newEntry && oldEntry.transportId === newEntry.transportId);
if (oldAlive && sameTransport) return;
-- rec._rolling = true;
-- const next = { mode: rec.mode, part: (rec.part || 1) + 1, baseTitle: rec.baseTitle };
-- console.log(`[VOD] Broadcaster re-produced video for stream ${streamId} on a new transport — rolling recording to part ${next.part}`);
-- this.stopRecording(streamId);
-- const timer = setTimeout(async () => {
-- try {
-- const live = await db.getStreamById(streamId);
-- if (!live || !live.is_live || this.isActivelyRecording(streamId)) return;
-- await this.startRecording(streamId, rec.protocol, {}, next);
-- } catch (e) { console.warn(`[VOD] roll restart failed for stream ${streamId}:`, e.message); }
-- }, 1500);
-- if (timer.unref) timer.unref();
++ this._rollRecording(streamId, rec, 'the broadcaster re-produced video on a new transport');
};
webrtcSFU.on('producer-added', this._producerWatch);
}
++
++ /** Stop a WebRTC recording and start its next part a moment later, on the producers there are then. */
++ _rollRecording(streamId, rec, why) {
++ rec._rolling = true;
++ const next = { mode: rec.mode, part: (rec.part || 1) + 1, baseTitle: rec.baseTitle };
++ console.log(`[VOD] Stream ${streamId}: ${why} — rolling recording to part ${next.part}`);
++ this.stopRecording(streamId);
++ const timer = setTimeout(async () => {
++ try {
++ const live = await db.getStreamById(streamId);
++ if (!live || !live.is_live || this.isActivelyRecording(streamId)) return;
++ await this.startRecording(streamId, rec.protocol, {}, next);
++ } catch (e) { console.warn(`[VOD] roll restart failed for stream ${streamId}:`, e.message); }
++ }, 1500);
++ if (timer.unref) timer.unref();
++ }
++
async startRecording(streamId, protocol, endpoint = {}, opts = {}) {
this._installProducerWatch();
if (this.activeRecordings.has(streamId)) {
@@@ -144,6 +161,7 @@ class StreamRecorder {
});
}
++ static get AUDIO_WAIT_MS() { return Number(process.env.VOD_AUDIO_WAIT_MS) || 5000; }
static get DISK_RETRY_DELAY_MS() { return 5 * 60 * 1000; }
static get DISK_RETRY_MAX() { return 12; } // ≈ 1 h of a still-live stream
@@@ -222,7 +240,17 @@ class StreamRecorder {
}
if (rec._cancel || !this.activeRecordings.has(streamId)) return;
rec.videoProducerId = videoProducer.id;
-- const audioProducer = webrtcSFU.findProducerByKind(roomId, 'audio');
++ // A browser produces its audio a moment after its video. Asked once, the instant the video arrived, the answer
++ // was "no audio" and the whole recording was silent (VOD 5986, 2026-10-25: both producers in the same second).
++ // So wait a few seconds for it: a stream with no microphone then records video only, and a microphone that
++ // comes later rolls the recording to a part with sound (_installProducerWatch).
++ let audioProducer = webrtcSFU.findProducerByKind(roomId, 'audio');
++ if (!audioProducer) {
++ try { audioProducer = await webrtcSFU.waitForProducer(roomId, 'audio', StreamRecorder.AUDIO_WAIT_MS); } catch { audioProducer = null; }
++ if (rec._cancel || !this.activeRecordings.has(streamId)) return;
++ }
++ rec.audioProducerId = audioProducer ? audioProducer.id : null;
++ rec.audioSettled = true;
// Codec info for Media's SDP: mimeType/clockRate from the producer; payload type
// from the router's preferred PT (that's what the PlainRTP consumer will emit).
ddiff --git a/test/recorder-audio.test.js b/test/recorder-audio.test.js
nnew file mode 100644
iindex 00000000..073c5faa
---- /dev/null
++++ b/test/recorder-audio.test.js
@@@ -0,0 +1,107 @@
++'use strict';
++// A browser stream's recording has its sound (server/streaming/recorder.js _startRtpRecording, _installProducerWatch).
++// VOD 5986 (goosely, 2026-10-25) was silent: the recorder found the video producer and asked for the audio one in the
++// same instant, a moment before the browser produced it, and recorded video only for seven minutes.
++// - Audio a moment after video: the recording waits for it and has sound.
++// - No microphone at all: after the wait it records video only.
++// - A microphone that comes later: the recording rolls to a new part that has the sound.
++// - A second microphone beside a live one, or the recorded one again: nothing happens.
++
++const assert = require('assert');
++const path = require('path');
++const { EventEmitter } = require('events');
++
++process.env.VOD_AUDIO_WAIT_MS = '300';
++
++// A small SFU: rooms of producers, the real findProducerByKind/waitForProducer behaviour, plain consumers recorded.
++const sfu = new EventEmitter();
++sfu.rooms = new Map();
++sfu.consumers = [];
++let nextId = 0;
++sfu.produce = (roomId, kind, transportId = 't1') => {
++ if (!sfu.rooms.has(roomId)) sfu.rooms.set(roomId, { producers: new Map(), router: { rtpCapabilities: { codecs: [] } } });
++ const id = `${kind}-${++nextId}`;
++ const producer = { id, kind, closed: false, rtpParameters: { codecs: [kind === 'video' ? { mimeType: 'video/VP8', payloadType: 96, clockRate: 90000 } : { mimeType: 'audio/opus', payloadType: 111, clockRate: 48000, channels: 2 }] } };
++ sfu.rooms.get(roomId).producers.set(id, { producer, transportId });
++ sfu.emit('producer-added', { roomId, producerId: id, kind });
++ return producer;
++};
++sfu.findProducerByKind = (roomId, kind) => {
++ const room = sfu.rooms.get(roomId);
++ if (!room) return null;
++ for (const [id, e] of [...room.producers.entries()].reverse()) if (e.producer.kind === kind && !e.producer.closed) return { id };
++ return null;
++};
++sfu.waitForProducer = (roomId, kind, timeoutMs) => {
++ const existing = sfu.findProducerByKind(roomId, kind);
++ if (existing) return Promise.resolve(existing);
++ return new Promise((resolve, reject) => {
++ const handler = (ev) => { if (ev.roomId === roomId && ev.kind === kind) { clearTimeout(timer); sfu.removeListener('producer-added', handler); resolve({ id: ev.producerId }); } };
++ const timer = setTimeout(() => { sfu.removeListener('producer-added', handler); reject(new Error('timeout')); }, timeoutMs);
++ sfu.on('producer-added', handler);
++ });
++};
++sfu.createPlainConsumer = async (roomId, producerId, ip, port) => { sfu.consumers.push({ roomId, producerId, port }); return { transportId: `plain-${producerId}`, payloadType: producerId.startsWith('video') ? 96 : 111 }; };
++sfu.closePlainConsumer = () => {};
++require.cache[path.resolve(__dirname, '../server/streaming/webrtc-sfu.js')] = { id: 'sfu', filename: 'sfu', loaded: true, exports: sfu };
++
++const db = require('../server/db/database');
++const media = require('../server/media-client');
++const recorder = require('../server/streaming/recorder');
++db.getStreamById = async (id) => ({ id, user_id: 1, title: 'Seattle', is_live: 1 });
++db.resolveStreamVodVisibility = async () => 'public';
++let vodSeq = 6000;
++const ingests = [];
++media.createVod = async () => ({ id: ++vodSeq });
++media.ingestRtpStart = async (vodId, { video, audio }) => { ingests.push({ vodId, video, audio }); return { videoPort: 12000, audioPort: audio ? 12002 : null }; };
++media.ingestRtpStop = async () => {};
++media.finalizeVod = async () => {};
++const wait = (ms) => new Promise((r) => setTimeout(r, ms));
++const until = async (fn, ms = 3000) => { const end = Date.now() + ms; while (!fn()) { if (Date.now() > end) throw new Error('timed out'); await wait(20); } };
++
++(async () => {
++ // 1. VOD 5986's race: the recording is waiting for video; audio comes just after it.
++ recorder.startRecording(101, 'webrtc', {}, { mode: 'vod' });
++ await wait(20);
++ sfu.produce('stream-101', 'video');
++ setTimeout(() => sfu.produce('stream-101', 'audio'), 30);
++ await until(() => ingests.length === 1);
++ assert.ok(ingests[0].audio && ingests[0].audio.codec === 'opus', 'the recording waited for the microphone and has sound');
++ assert.deepStrictEqual(sfu.consumers.filter((c) => c.roomId === 'stream-101').map((c) => c.producerId.split('-')[0]).sort(), ['audio', 'video']);
++ console.log('✅ audio a moment after video is recorded');
++
++ // A second microphone beside the live one, and the recorded one again: nothing happens.
++ sfu.produce('stream-101', 'audio');
++ sfu.emit('producer-added', { roomId: 'stream-101', producerId: recorder.activeRecordings.get(101).audioProducerId, kind: 'audio' });
++ await wait(50);
++ assert.strictEqual(recorder.activeRecordings.get(101).part, 1, 'no roll for a second microphone beside a live one');
++ console.log('✅ a second microphone beside a live one is not followed');
++
++ // 2. No microphone: video only after the wait.
++ sfu.produce('stream-102', 'video');
++ recorder.startRecording(102, 'webrtc', {}, { mode: 'vod' });
++ await until(() => ingests.length === 2, 2000);
++ assert.strictEqual(ingests[1].audio, undefined, 'a stream with no microphone records video only');
++ console.log('✅ no microphone: video only after a short wait');
++
++ // 3. The microphone comes later: the recording rolls to a part with sound.
++ const before = recorder.activeRecordings.get(102);
++ assert.strictEqual(before.part, 1);
++ sfu.produce('stream-102', 'audio');
++ await until(() => ingests.length === 3, 4000);
++ assert.ok(ingests[2].audio, 'the next part has the sound');
++ assert.strictEqual(recorder.activeRecordings.get(102).part, 2);
++ console.log('✅ a late microphone rolls the recording to a part with sound');
++
++ // 4. The recorded microphone dies and a new one comes (same camera): roll again.
++ const rec = recorder.activeRecordings.get(102);
++ sfu.rooms.get('stream-102').producers.get(rec.audioProducerId).producer.closed = true;
++ sfu.produce('stream-102', 'audio');
++ await until(() => ingests.length === 4, 4000);
++ assert.strictEqual(recorder.activeRecordings.get(102).part, 3);
++ console.log('✅ a new microphone after the recorded one died rolls too');
++
++ for (const id of [101, 102]) recorder.stopRecording(id);
++ console.log('\n✅ recorder audio: waits for the microphone, video-only without one, rolls when one comes later');
++ process.exit(0);
++})().catch((err) => { console.error(err); process.exit(1); });