openvibers/openvibe.live Commits
Timeline: stop orphaning spools on restart, stop dropping stream tails
Changes
@@@ -309,13 +309,16 @@ function stopAllCaptures() {* Each entry carries the absolute offset (seconds into the stream) its audio starts at,* derived from the segment index, so timeline timestamps stay correct.*/--function pendingSegments(streamId) {++function pendingSegments(streamId, { includeLast = false } = {}) {const dir = spoolDir(streamId);let names;try { names = fs.readdirSync(dir).filter(n => /^seg-\d{6}\.wav$/.test(n)).sort(); }catch { return []; }-- if (names.length <= 1) return []; // newest is still being written-- const ready = names.slice(0, -1);++ // Normally the newest file is still being written by ffmpeg, so it is not eligible.++ // Once capture has STOPPED there is no writer, so the final segment is complete and++ // must be included — otherwise the last seconds of every stream are silently dropped.++ if (!includeLast && names.length <= 1) return [];++ const ready = includeLast ? names : names.slice(0, -1);// Drop the oldest if the spool has run away (slow transcriber / very long stream).const overflow = Math.max(0, ready.length - MAX_SPOOL_FILES);for (let i = 0; i < overflow; i++) {@@@ -340,6 +343,15 @@ function discardSegment(streamId, name) {try { fs.unlinkSync(path.join(spoolDir(streamId), name)); return true; } catch { return false; }}++/** Stream ids that currently have a spool directory on disk, live or not. */++function spooledStreamIds() {++ try {++ return fs.readdirSync(SPOOL_ROOT, { withFileTypes: true })++ .filter(d => d.isDirectory() && /^\d+$/.test(d.name))++ .map(d => parseInt(d.name, 10));++ } catch { return []; }++}++/** Remove a stream's whole spool directory (call when the stream ends). */function purgeSpool(streamId) {try { fs.rmSync(spoolDir(streamId), { recursive: true, force: true }); return true; }@@@ -349,6 +361,6 @@ function purgeSpool(streamId) {module.exports = {captureAudioChunk,startContinuousCapture, stopContinuousCapture, stopAllCaptures, isCapturing,-- pendingSegments, discardSegment, purgeSpool, spoolDir,++ pendingSegments, discardSegment, purgeSpool, spoolDir, spooledStreamIds,SEGMENT_SEC,};@@@ -127,15 +127,27 @@ async function tick() {_known.add(stream.id);}-- // Stop + tidy streams that ended. Their remaining segments are drained first so-- // the tail of the stream is not lost, then the spool is removed.-- for (const id of [..._known]) {++ // Stop + tidy streams that ended. Scan the SPOOL DIRECTORY rather than the in-memory++ // _known set: _known is empty after a restart, so any stream that ended while the++ // service was down would leave its directory behind forever. Six such orphans were++ // found on disk from earlier restarts.++ //++ // includeLast matters here too. While capture is running the newest file is still++ // being written, so it is skipped — but once ffmpeg is gone that file is complete,++ // and skipping it would silently drop the last seconds of every stream.++ for (const id of audio.spooledStreamIds()) {if (liveIds.has(id)) continue;audio.stopContinuousCapture(id);-- const leftovers = audio.pendingSegments(id);++ const leftovers = audio.pendingSegments(id, { includeLast: true });if (!leftovers.length) {audio.purgeSpool(id);_known.delete(id);++ continue;++ }++ // Drain the tail a few segments per tick, same as for a live stream.++ const stream = { id, user_id: null };++ for (const seg of leftovers.slice(0, MAX_SEGMENTS_PER_TICK)) {++ await _processSegment(stream, seg).catch(() => {});}}
@@ -309,13 +309,16 @@ function stopAllCaptures() {
@@ -309,13 +309,16 @@ function stopAllCaptures() {
* Each entry carries the absolute offset (seconds into the stream) its audio starts at,
* Each entry carries the absolute offset (seconds into the stream) its audio starts at,
* derived from the segment index, so timeline timestamps stay correct.
* derived from the segment index, so timeline timestamps stay correct.
*/
*/
-function pendingSegments(streamId) {
+function pendingSegments(streamId, { includeLast = false } = {}) {
const dir = spoolDir(streamId);
const dir = spoolDir(streamId);
let names;
let names;
try { names = fs.readdirSync(dir).filter(n => /^seg-\d{6}\.wav$/.test(n)).sort(); }
try { names = fs.readdirSync(dir).filter(n => /^seg-\d{6}\.wav$/.test(n)).sort(); }
catch { return []; }
catch { return []; }
- if (names.length <= 1) return []; // newest is still being written
- const ready = names.slice(0, -1);
+ // Normally the newest file is still being written by ffmpeg, so it is not eligible.
+ // Once capture has STOPPED there is no writer, so the final segment is complete and
+ // must be included — otherwise the last seconds of every stream are silently dropped.
+ if (!includeLast && names.length <= 1) return [];
+ const ready = includeLast ? names : names.slice(0, -1);
// Drop the oldest if the spool has run away (slow transcriber / very long stream).
// Drop the oldest if the spool has run away (slow transcriber / very long stream).
const overflow = Math.max(0, ready.length - MAX_SPOOL_FILES);
const overflow = Math.max(0, ready.length - MAX_SPOOL_FILES);
for (let i = 0; i < overflow; i++) {
for (let i = 0; i < overflow; i++) {
@@ -340,6 +343,15 @@ function discardSegment(streamId, name) {
@@ -340,6 +343,15 @@ function discardSegment(streamId, name) {
try { fs.unlinkSync(path.join(spoolDir(streamId), name)); return true; } catch { return false; }
try { fs.unlinkSync(path.join(spoolDir(streamId), name)); return true; } catch { return false; }
}
}
+/** Stream ids that currently have a spool directory on disk, live or not. */
+function spooledStreamIds() {
+ try {
+ return fs.readdirSync(SPOOL_ROOT, { withFileTypes: true })
+ .filter(d => d.isDirectory() && /^\d+$/.test(d.name))
+ .map(d => parseInt(d.name, 10));
+ } catch { return []; }
+}
+
/** Remove a stream's whole spool directory (call when the stream ends). */
/** Remove a stream's whole spool directory (call when the stream ends). */
function purgeSpool(streamId) {
function purgeSpool(streamId) {
try { fs.rmSync(spoolDir(streamId), { recursive: true, force: true }); return true; }
try { fs.rmSync(spoolDir(streamId), { recursive: true, force: true }); return true; }
@@ -349,6 +361,6 @@ function purgeSpool(streamId) {
@@ -349,6 +361,6 @@ function purgeSpool(streamId) {
module.exports = {
module.exports = {
captureAudioChunk,
captureAudioChunk,
startContinuousCapture, stopContinuousCapture, stopAllCaptures, isCapturing,
startContinuousCapture, stopContinuousCapture, stopAllCaptures, isCapturing,
- pendingSegments, discardSegment, purgeSpool, spoolDir,
+ pendingSegments, discardSegment, purgeSpool, spoolDir, spooledStreamIds,
SEGMENT_SEC,
SEGMENT_SEC,
};
};
@@ -127,15 +127,27 @@ async function tick() {
@@ -127,15 +127,27 @@ async function tick() {
_known.add(stream.id);
_known.add(stream.id);
}
}
- // Stop + tidy streams that ended. Their remaining segments are drained first so
- // the tail of the stream is not lost, then the spool is removed.
- for (const id of [..._known]) {
+ // Stop + tidy streams that ended. Scan the SPOOL DIRECTORY rather than the in-memory
+ // _known set: _known is empty after a restart, so any stream that ended while the
+ // service was down would leave its directory behind forever. Six such orphans were
+ // found on disk from earlier restarts.
+ //
+ // includeLast matters here too. While capture is running the newest file is still
+ // being written, so it is skipped — but once ffmpeg is gone that file is complete,
+ // and skipping it would silently drop the last seconds of every stream.
+ for (const id of audio.spooledStreamIds()) {
if (liveIds.has(id)) continue;
if (liveIds.has(id)) continue;
audio.stopContinuousCapture(id);
audio.stopContinuousCapture(id);
- const leftovers = audio.pendingSegments(id);
+ const leftovers = audio.pendingSegments(id, { includeLast: true });
if (!leftovers.length) {
if (!leftovers.length) {
audio.purgeSpool(id);
audio.purgeSpool(id);
_known.delete(id);
_known.delete(id);
+ continue;
+ }
+ // Drain the tail a few segments per tick, same as for a live stream.
+ const stream = { id, user_id: null };
+ for (const seg of leftovers.slice(0, MAX_SEGMENTS_PER_TICK)) {
+ await _processSegment(stream, seg).catch(() => {});
}
}
}
}