openvibers/openvibe.live Commits
Live: files and proxied bodies go to responses through stream.pipeline (deleted live thumbnails stayed open) (#117)
Changes
@@@ -130,7 +130,7 @@ router.get('/voice/:user', optionalAuth, async (req, res) => {if (who !== 'announcer' && !user) return res.status(404).json({ error: 'No such user' });const out = await voice.speak({ user, text });res.set({ 'Content-Type': out.mimeType, 'Cache-Control': 'public, max-age=604800, immutable', 'X-Arena-Voice': out.voice, 'X-Cache': out.cached ? 'HIT' : 'MISS' });-- require('fs').createReadStream(out.path).pipe(res);++ require('stream').pipeline(require('fs').createReadStream(out.path), res, () => {}); // closes the file when the listener leaves} catch (err) { res.status(err.message && /budget|hoarse/.test(err.message) ? 429 : 500).json({ error: err.message || 'Voice failed' }); }});@@@ -245,7 +245,7 @@ contextRouter.get('/tts-audio/:file', (req, res) => {const hit = require('../arena/voice').cachedByName(req.params.file);if (!hit) return fail(res, 404, 'No such clip');res.set({ 'Content-Type': hit.mimeType });-- require('fs').createReadStream(hit.path).pipe(res);++ require('stream').pipeline(require('fs').createReadStream(hit.path), res, () => {}); // closes the file when the listener leaves} catch (err) { fail(res, 500, err.message); }});@@@ -382,8 +382,9 @@ async function proxy(req, res, apiPath, { actingUser, method, query, body } = {}if (v) res.set(h, v);}if (!upstream.body) return res.end();-- const { Readable } = require('stream');-- Readable.fromWeb(upstream.body).pipe(res);++ const { Readable, pipeline } = require('stream');++ // pipeline: a viewer who leaves cancels Media's body too (.pipe() kept downloading it into a closed response).++ pipeline(Readable.fromWeb(upstream.body), res, () => {});}/**@@@ -10,6 +10,9 @@const express = require('express');const path = require('path');const fs = require('fs');++// pipeline, never .pipe(res): a viewer who leaves mid-file destroys the response, and .pipe() left the file open for the++// life of the process (36 deleted live thumbnails were still open after 12 h, 2026-10-13).++const { pipeline } = require('stream');const multer = require('multer');const db = require('../db/database');const { can } = require('../auth/permissions');@@@ -134,7 +137,7 @@ router.get('/:filename', async (req, res) => {const current = await liveThumbs.getStreamThumbnailState(Number(alias[1]));if (current.exists) {res.writeHead(200, { 'Content-Type': 'image/jpeg', 'Content-Length': fs.statSync(current.filePath).size, 'Cache-Control': 'no-cache' });-- return fs.createReadStream(current.filePath).pipe(res);++ return pipeline(fs.createReadStream(current.filePath), res, () => {});}res.set('Cache-Control', 'no-cache');// The fallback frame for a WebRTC stream is a Media thumbnail URL (server/index.js).@@@ -149,7 +152,7 @@ router.get('/:filename', async (req, res) => {'Content-Length': stat.size,'Cache-Control': 'public, max-age=30',});-- return fs.createReadStream(localPath).pipe(res);++ return pipeline(fs.createReadStream(localPath), res, () => {});}if (filename.startsWith('stream-')) {// Missing live thumbnail → transparent pixel (old behavior; avoids broken cards).@@@ -137,7 +137,7 @@ function proxy() {delete headers['proxy-connection']; delete headers['proxy-authorization'];const upstream = http.request(target, { method: req.method, headers, lookup: safeLookup }, (up) => {res.writeHead(up.statusCode, up.headers);-- up.pipe(res);++ require('stream').pipeline(up, res, () => {}); // a client that leaves ends the upstream request too});upstream.on('error', (e) => { if (!res.headersSent) res.writeHead(e.code === 'EGRESS_DENIED' ? 403 : 502); res.end(); });req.pipe(upstream);nnew file mode 100644@@@ -0,0 +1,30 @@++'use strict';++// A file or an upstream body sent to a response goes through stream.pipeline, never .pipe(res): when the viewer leaves,++// the response is destroyed and .pipe() does not destroy its source, so the file stayed open for the life of the process++// (36 deleted live thumbnails still open after 12 h, 2026-10-13) and a proxied body kept downloading into nothing.++const assert = require('assert');++const fs = require('fs');++const path = require('path');++++const SERVER = path.join(__dirname, '..', 'server');++const files = [];++(function walk(dir) {++ for (const e of fs.readdirSync(dir, { withFileTypes: true })) {++ const p = path.join(dir, e.name);++ if (e.isDirectory()) walk(p);++ else if (e.name.endsWith('.js')) files.push(p);++ }++})(SERVER);++++const offenders = [];++for (const f of files) {++ fs.readFileSync(f, 'utf8').split('\n').forEach((line, i) => {++ if (/^\s*(\/\/|\*)/.test(line)) return;++ if (/createReadStream\([^)]*\)\s*\.pipe\(/.test(line) || /fromWeb\([^)]*\)\s*\.pipe\(/.test(line)) offenders.push(`${path.relative(SERVER, f)}:${i + 1}`);++ });++}++assert.deepStrictEqual(offenders, [], `use stream.pipeline(source, res, cb): ${offenders.join(', ')}`);++++const thumbs = fs.readFileSync(path.join(SERVER, 'media-proxy', 'thumbnails.js'), 'utf8');++assert.strictEqual((thumbs.match(/pipeline\(fs\.createReadStream\(/g) || []).length, 2, 'both live-thumbnail responses use pipeline');++console.log(`stream-pipeline: ${files.length} server files checked`);
@@ -130,7 +130,7 @@ router.get('/voice/:user', optionalAuth, async (req, res) => {
@@ -130,7 +130,7 @@ router.get('/voice/:user', optionalAuth, async (req, res) => {
if (who !== 'announcer' && !user) return res.status(404).json({ error: 'No such user' });
if (who !== 'announcer' && !user) return res.status(404).json({ error: 'No such user' });
const out = await voice.speak({ user, text });
const out = await voice.speak({ user, text });
res.set({ 'Content-Type': out.mimeType, 'Cache-Control': 'public, max-age=604800, immutable', 'X-Arena-Voice': out.voice, 'X-Cache': out.cached ? 'HIT' : 'MISS' });
res.set({ 'Content-Type': out.mimeType, 'Cache-Control': 'public, max-age=604800, immutable', 'X-Arena-Voice': out.voice, 'X-Cache': out.cached ? 'HIT' : 'MISS' });
- require('fs').createReadStream(out.path).pipe(res);
+ require('stream').pipeline(require('fs').createReadStream(out.path), res, () => {}); // closes the file when the listener leaves
} catch (err) { res.status(err.message && /budget|hoarse/.test(err.message) ? 429 : 500).json({ error: err.message || 'Voice failed' }); }
} catch (err) { res.status(err.message && /budget|hoarse/.test(err.message) ? 429 : 500).json({ error: err.message || 'Voice failed' }); }
});
});
@@ -245,7 +245,7 @@ contextRouter.get('/tts-audio/:file', (req, res) => {
@@ -245,7 +245,7 @@ contextRouter.get('/tts-audio/:file', (req, res) => {
const hit = require('../arena/voice').cachedByName(req.params.file);
const hit = require('../arena/voice').cachedByName(req.params.file);
if (!hit) return fail(res, 404, 'No such clip');
if (!hit) return fail(res, 404, 'No such clip');
res.set({ 'Content-Type': hit.mimeType });
res.set({ 'Content-Type': hit.mimeType });
- require('fs').createReadStream(hit.path).pipe(res);
+ require('stream').pipeline(require('fs').createReadStream(hit.path), res, () => {}); // closes the file when the listener leaves
} catch (err) { fail(res, 500, err.message); }
} catch (err) { fail(res, 500, err.message); }
});
});
@@ -382,8 +382,9 @@ async function proxy(req, res, apiPath, { actingUser, method, query, body } = {}
@@ -382,8 +382,9 @@ async function proxy(req, res, apiPath, { actingUser, method, query, body } = {}
if (v) res.set(h, v);
if (v) res.set(h, v);
}
}
if (!upstream.body) return res.end();
if (!upstream.body) return res.end();
- const { Readable } = require('stream');
- Readable.fromWeb(upstream.body).pipe(res);
+ const { Readable, pipeline } = require('stream');
+ // pipeline: a viewer who leaves cancels Media's body too (.pipe() kept downloading it into a closed response).
+ pipeline(Readable.fromWeb(upstream.body), res, () => {});
}
}
/**
/**
@@ -10,6 +10,9 @@
@@ -10,6 +10,9 @@
const express = require('express');
const express = require('express');
const path = require('path');
const path = require('path');
const fs = require('fs');
const fs = require('fs');
+// pipeline, never .pipe(res): a viewer who leaves mid-file destroys the response, and .pipe() left the file open for the
+// life of the process (36 deleted live thumbnails were still open after 12 h, 2026-10-13).
+const { pipeline } = require('stream');
const multer = require('multer');
const multer = require('multer');
const db = require('../db/database');
const db = require('../db/database');
const { can } = require('../auth/permissions');
const { can } = require('../auth/permissions');
@@ -134,7 +137,7 @@ router.get('/:filename', async (req, res) => {
@@ -134,7 +137,7 @@ router.get('/:filename', async (req, res) => {
const current = await liveThumbs.getStreamThumbnailState(Number(alias[1]));
const current = await liveThumbs.getStreamThumbnailState(Number(alias[1]));
if (current.exists) {
if (current.exists) {
res.writeHead(200, { 'Content-Type': 'image/jpeg', 'Content-Length': fs.statSync(current.filePath).size, 'Cache-Control': 'no-cache' });
res.writeHead(200, { 'Content-Type': 'image/jpeg', 'Content-Length': fs.statSync(current.filePath).size, 'Cache-Control': 'no-cache' });
- return fs.createReadStream(current.filePath).pipe(res);
+ return pipeline(fs.createReadStream(current.filePath), res, () => {});
}
}
res.set('Cache-Control', 'no-cache');
res.set('Cache-Control', 'no-cache');
// The fallback frame for a WebRTC stream is a Media thumbnail URL (server/index.js).
// The fallback frame for a WebRTC stream is a Media thumbnail URL (server/index.js).
@@ -149,7 +152,7 @@ router.get('/:filename', async (req, res) => {
@@ -149,7 +152,7 @@ router.get('/:filename', async (req, res) => {
'Content-Length': stat.size,
'Content-Length': stat.size,
'Cache-Control': 'public, max-age=30',
'Cache-Control': 'public, max-age=30',
});
});
- return fs.createReadStream(localPath).pipe(res);
+ return pipeline(fs.createReadStream(localPath), res, () => {});
}
}
if (filename.startsWith('stream-')) {
if (filename.startsWith('stream-')) {
// Missing live thumbnail → transparent pixel (old behavior; avoids broken cards).
// Missing live thumbnail → transparent pixel (old behavior; avoids broken cards).
@@ -137,7 +137,7 @@ function proxy() {
@@ -137,7 +137,7 @@ function proxy() {
delete headers['proxy-connection']; delete headers['proxy-authorization'];
delete headers['proxy-connection']; delete headers['proxy-authorization'];
const upstream = http.request(target, { method: req.method, headers, lookup: safeLookup }, (up) => {
const upstream = http.request(target, { method: req.method, headers, lookup: safeLookup }, (up) => {
res.writeHead(up.statusCode, up.headers);
res.writeHead(up.statusCode, up.headers);
- up.pipe(res);
+ require('stream').pipeline(up, res, () => {}); // a client that leaves ends the upstream request too
});
});
upstream.on('error', (e) => { if (!res.headersSent) res.writeHead(e.code === 'EGRESS_DENIED' ? 403 : 502); res.end(); });
upstream.on('error', (e) => { if (!res.headersSent) res.writeHead(e.code === 'EGRESS_DENIED' ? 403 : 502); res.end(); });
req.pipe(upstream);
req.pipe(upstream);
new file mode 100644
new file mode 100644
@@ -0,0 +1,30 @@
@@ -0,0 +1,30 @@
+'use strict';
+// A file or an upstream body sent to a response goes through stream.pipeline, never .pipe(res): when the viewer leaves,
+// the response is destroyed and .pipe() does not destroy its source, so the file stayed open for the life of the process
+// (36 deleted live thumbnails still open after 12 h, 2026-10-13) and a proxied body kept downloading into nothing.
+const assert = require('assert');
+const fs = require('fs');
+const path = require('path');
+
+const SERVER = path.join(__dirname, '..', 'server');
+const files = [];
+(function walk(dir) {
+ for (const e of fs.readdirSync(dir, { withFileTypes: true })) {
+ const p = path.join(dir, e.name);
+ if (e.isDirectory()) walk(p);
+ else if (e.name.endsWith('.js')) files.push(p);
+ }
+})(SERVER);
+
+const offenders = [];
+for (const f of files) {
+ fs.readFileSync(f, 'utf8').split('\n').forEach((line, i) => {
+ if (/^\s*(\/\/|\*)/.test(line)) return;
+ if (/createReadStream\([^)]*\)\s*\.pipe\(/.test(line) || /fromWeb\([^)]*\)\s*\.pipe\(/.test(line)) offenders.push(`${path.relative(SERVER, f)}:${i + 1}`);
+ });
+}
+assert.deepStrictEqual(offenders, [], `use stream.pipeline(source, res, cb): ${offenders.join(', ')}`);
+
+const thumbs = fs.readFileSync(path.join(SERVER, 'media-proxy', 'thumbnails.js'), 'utf8');
+assert.strictEqual((thumbs.match(/pipeline\(fs\.createReadStream\(/g) || []).length, 2, 'both live-thumbnail responses use pipeline');
+console.log(`stream-pipeline: ${files.length} server files checked`);