openvibers/openvibe.media Commits
Two uploads/thumbnail paths no longer hang on a database error: a multipart part's completion read (getSession) sat outside its try, so a rejected read was an unhandled rejection and the PUT never answered; it now answers 500 media.object.store_failed. A thumbnail probe whose close handler throws resolves the request (null thumbnail) instead of leaving it unsettled. Tests: test/multipart-part-db-error.test.js, test/thumbnail-probe-throw.test.js. Found by the estate-wide floating-promise sweep; fixed by DeepSeek, reviewed independently (VERDICT: SHIP). npm test 67/68, 1 skipped (N-1 SQL across the PostgreSQL switch). npm test 67/68 test files passed, 1 skipped (n-1.test.js: N-1 SQL: skipped (the recorded N-1 releas.
Changes
@@@ -126,10 +126,10 @@ function receivePart(req, session, n, { expectSha256 = null } = {}) {const sha = hash.digest('hex');if (got !== want) { drop(); return await finish({ status: 400, code: 'media.upload.part_size_mismatch', error: `Part ${n} must be ${want} bytes, received ${got}` }); }if (expectSha256 && String(expectSha256).toLowerCase() !== sha) { drop(); return await finish({ status: 400, code: 'media.upload.part_hash_mismatch', error: `Part ${n} does not match X-Content-SHA256` }); }-- // The session may have been completed or aborted while this part streamed in.-- const now = await getSession(session.id);-- if (!now || now.status !== 'active') { drop(); return await finish({ status: 409, code: 'media.upload.not_active', error: `The upload is ${now ? now.status : 'gone'}` }); }try {++ // The session may have been completed or aborted while this part streamed in.++ const now = await getSession(session.id);++ if (!now || now.status !== 'active') { drop(); return await finish({ status: 409, code: 'media.upload.not_active', error: `The upload is ${now ? now.status : 'gone'}` }); }fs.renameSync(tmp, partFile(session.id, n));await db.run(`INSERT INTO media_upload_parts (upload_id, part_number, size_bytes, sha256) VALUES (?, ?, ?, ?)ON CONFLICT(upload_id, part_number) DO UPDATE SET size_bytes = excluded.size_bytes, sha256 = excluded.sha256, received_at = ov_now()`,@@@ -118,7 +118,7 @@ function generateFromVideo(videoPath, prefix, entityId, opts = {}) {let probeData = '';probe.stdout.on('data', (d) => (probeData += d));-- probe.on('close', async (probeCode) => {++ const onProbeClose = async (probeCode) => {let seekTime = opts.seekSeconds || 1;if (!opts.seekSeconds) {@@@ -175,7 +175,9 @@ function generateFromVideo(videoPath, prefix, entityId, opts = {}) {}});ff.on('error', () => { clearTimeout(killTimer); resolve(null); });-- });++ };++ // A throw inside the handler (e.g. spawn failing) must resolve the request, not hang it.++ probe.on('close', (probeCode) => { onProbeClose(probeCode).catch(() => resolve(null)); });probe.on('error', () => resolve(null));}).finally(() => _activeThumbJobs.delete(jobKey));nnew file mode 100644@@@ -0,0 +1,44 @@++'use strict';++// The part-completion listener read the session row (getSession) outside its try, so a rejected read++// was an unhandled rejection — Media installs no global handler — and the returned promise never++// settled, so the part request hung. The read now sits inside the try: a failure answers 500.++const assert = require('assert');++const fs = require('fs');++const os = require('os');++const path = require('path');++const { PassThrough } = require('stream');++++(async () => {++ const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-mpdb-'));++ process.env.OBJECTS_PATH = path.join(tmp, 'objects');++ process.env.VOD_PATH = path.join(tmp, 'vods');++ process.env.FILES_PATH = path.join(tmp, 'files');++ process.env.THUMBNAILS_PATH = path.join(tmp, 'thumbs');++++ const db = require('../server/db/database');++ const multipart = require('../server/objects/multipart');++++ // A 1-part session of 4 bytes: the close listener hashes, checks the size, then reads the row.++ const session = { id: 'mup_test_dbdown', part_size: 4, total_size: 4, parts_expected: 1, status: 'active' };++ const origGet = db.get;++ let unhandled = null;++ const onUnhandled = (e) => { unhandled = e; };++ process.on('unhandledRejection', onUnhandled);++ db.get = async () => { throw new Error('database is down'); };++ try {++ const req = new PassThrough();++ const p = multipart.receivePart(req, session, 1);++ req.end(Buffer.alloc(4, 7));++ const result = await Promise.race([p, new Promise((r) => setTimeout(() => r('HUNG'), 3000))]);++ await new Promise((r) => setTimeout(r, 50)); // let a stray unhandled rejection surface++ assert.notStrictEqual(result, 'HUNG', 'the part request must settle when getSession rejects');++ assert.deepStrictEqual([result.status, result.code], [500, 'media.object.store_failed']);++ assert.strictEqual(unhandled, null, `unhandled rejection: ${unhandled && unhandled.message}`);++ } finally {++ db.get = origGet;++ process.off('unhandledRejection', onUnhandled);++ }++++ console.log('✅ multipart part completion: a rejected getSession answers 500, no unhandled rejection and no hang');++ process.exit(0);++})().catch((err) => { console.error(err); process.exit(1); });nnew file mode 100644@@@ -0,0 +1,50 @@++'use strict';++// generateFromVideo's promise had no reject path: a throw inside the async 'close' listener (here the++// ffmpeg spawn) left it forever unsettled, so the awaiting request hung, and the rejection was++// unhandled. The listener now catches and resolves null.++const assert = require('assert');++const fs = require('fs');++const os = require('os');++const path = require('path');++const { EventEmitter } = require('events');++++(async () => {++ const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-thumbthrow-'));++ process.env.THUMBNAILS_PATH = path.join(tmp, 'thumbs');++ process.env.VOD_PATH = path.join(tmp, 'vods');++ process.env.FILES_PATH = path.join(tmp, 'files');++++ // The probe's ffprobe child closes cleanly; the ffmpeg spawn that follows throws.++ const cp = require('child_process');++ const realSpawn = cp.spawn;++ cp.spawn = (cmd) => {++ if (cmd !== 'ffprobe') throw new Error(`cannot spawn ${cmd}`);++ const child = new EventEmitter();++ child.stdout = new EventEmitter();++ child.stderr = new EventEmitter();++ setImmediate(() => child.emit('close', 0));++ return child;++ };++ let unhandled = null;++ const onUnhandled = (e) => { unhandled = e; };++ process.on('unhandledRejection', onUnhandled);++++ const file = path.join(tmp, 'v.mp4');++ fs.writeFileSync(file, 'not really a video');++ try {++ const thumb = require('../server/thumbnails/thumbnail-service');++ // seekSeconds skips the probe-duration branch, so the close handler goes straight to ffmpeg.++ const p = thumb.generateFromVideo(file, 'vod', 4242, { seekSeconds: 1 });++ const result = await Promise.race([p, new Promise((r) => setTimeout(() => r('HUNG'), 3000))]);++ await new Promise((r) => setTimeout(r, 50)); // let a stray unhandled rejection surface++ assert.notStrictEqual(result, 'HUNG', 'the thumbnail request must settle when the close handler throws');++ assert.strictEqual(result, null);++ assert.strictEqual(unhandled, null, `unhandled rejection: ${unhandled && unhandled.message}`);++ } finally {++ cp.spawn = realSpawn;++ process.off('unhandledRejection', onUnhandled);++ }++++ console.log('✅ thumbnail probe: a throwing close handler resolves null instead of hanging');++ process.exit(0);++})().catch((err) => { console.error(err); process.exit(1); });
@@ -126,10 +126,10 @@ function receivePart(req, session, n, { expectSha256 = null } = {}) {
@@ -126,10 +126,10 @@ function receivePart(req, session, n, { expectSha256 = null } = {}) {
const sha = hash.digest('hex');
const sha = hash.digest('hex');
if (got !== want) { drop(); return await finish({ status: 400, code: 'media.upload.part_size_mismatch', error: `Part ${n} must be ${want} bytes, received ${got}` }); }
if (got !== want) { drop(); return await finish({ status: 400, code: 'media.upload.part_size_mismatch', error: `Part ${n} must be ${want} bytes, received ${got}` }); }
if (expectSha256 && String(expectSha256).toLowerCase() !== sha) { drop(); return await finish({ status: 400, code: 'media.upload.part_hash_mismatch', error: `Part ${n} does not match X-Content-SHA256` }); }
if (expectSha256 && String(expectSha256).toLowerCase() !== sha) { drop(); return await finish({ status: 400, code: 'media.upload.part_hash_mismatch', error: `Part ${n} does not match X-Content-SHA256` }); }
- // The session may have been completed or aborted while this part streamed in.
- const now = await getSession(session.id);
- if (!now || now.status !== 'active') { drop(); return await finish({ status: 409, code: 'media.upload.not_active', error: `The upload is ${now ? now.status : 'gone'}` }); }
try {
try {
+ // The session may have been completed or aborted while this part streamed in.
+ const now = await getSession(session.id);
+ if (!now || now.status !== 'active') { drop(); return await finish({ status: 409, code: 'media.upload.not_active', error: `The upload is ${now ? now.status : 'gone'}` }); }
fs.renameSync(tmp, partFile(session.id, n));
fs.renameSync(tmp, partFile(session.id, n));
await db.run(`INSERT INTO media_upload_parts (upload_id, part_number, size_bytes, sha256) VALUES (?, ?, ?, ?)
await db.run(`INSERT INTO media_upload_parts (upload_id, part_number, size_bytes, sha256) VALUES (?, ?, ?, ?)
ON CONFLICT(upload_id, part_number) DO UPDATE SET size_bytes = excluded.size_bytes, sha256 = excluded.sha256, received_at = ov_now()`,
ON CONFLICT(upload_id, part_number) DO UPDATE SET size_bytes = excluded.size_bytes, sha256 = excluded.sha256, received_at = ov_now()`,
@@ -118,7 +118,7 @@ function generateFromVideo(videoPath, prefix, entityId, opts = {}) {
@@ -118,7 +118,7 @@ function generateFromVideo(videoPath, prefix, entityId, opts = {}) {
let probeData = '';
let probeData = '';
probe.stdout.on('data', (d) => (probeData += d));
probe.stdout.on('data', (d) => (probeData += d));
- probe.on('close', async (probeCode) => {
+ const onProbeClose = async (probeCode) => {
let seekTime = opts.seekSeconds || 1;
let seekTime = opts.seekSeconds || 1;
if (!opts.seekSeconds) {
if (!opts.seekSeconds) {
@@ -175,7 +175,9 @@ function generateFromVideo(videoPath, prefix, entityId, opts = {}) {
@@ -175,7 +175,9 @@ function generateFromVideo(videoPath, prefix, entityId, opts = {}) {
}
}
});
});
ff.on('error', () => { clearTimeout(killTimer); resolve(null); });
ff.on('error', () => { clearTimeout(killTimer); resolve(null); });
- });
+ };
+ // A throw inside the handler (e.g. spawn failing) must resolve the request, not hang it.
+ probe.on('close', (probeCode) => { onProbeClose(probeCode).catch(() => resolve(null)); });
probe.on('error', () => resolve(null));
probe.on('error', () => resolve(null));
}).finally(() => _activeThumbJobs.delete(jobKey));
}).finally(() => _activeThumbJobs.delete(jobKey));
new file mode 100644
new file mode 100644
@@ -0,0 +1,44 @@
@@ -0,0 +1,44 @@
+'use strict';
+// The part-completion listener read the session row (getSession) outside its try, so a rejected read
+// was an unhandled rejection — Media installs no global handler — and the returned promise never
+// settled, so the part request hung. The read now sits inside the try: a failure answers 500.
+const assert = require('assert');
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+const { PassThrough } = require('stream');
+
+(async () => {
+ const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-mpdb-'));
+ process.env.OBJECTS_PATH = path.join(tmp, 'objects');
+ process.env.VOD_PATH = path.join(tmp, 'vods');
+ process.env.FILES_PATH = path.join(tmp, 'files');
+ process.env.THUMBNAILS_PATH = path.join(tmp, 'thumbs');
+
+ const db = require('../server/db/database');
+ const multipart = require('../server/objects/multipart');
+
+ // A 1-part session of 4 bytes: the close listener hashes, checks the size, then reads the row.
+ const session = { id: 'mup_test_dbdown', part_size: 4, total_size: 4, parts_expected: 1, status: 'active' };
+ const origGet = db.get;
+ let unhandled = null;
+ const onUnhandled = (e) => { unhandled = e; };
+ process.on('unhandledRejection', onUnhandled);
+ db.get = async () => { throw new Error('database is down'); };
+ try {
+ const req = new PassThrough();
+ const p = multipart.receivePart(req, session, 1);
+ req.end(Buffer.alloc(4, 7));
+ const result = await Promise.race([p, new Promise((r) => setTimeout(() => r('HUNG'), 3000))]);
+ await new Promise((r) => setTimeout(r, 50)); // let a stray unhandled rejection surface
+ assert.notStrictEqual(result, 'HUNG', 'the part request must settle when getSession rejects');
+ assert.deepStrictEqual([result.status, result.code], [500, 'media.object.store_failed']);
+ assert.strictEqual(unhandled, null, `unhandled rejection: ${unhandled && unhandled.message}`);
+ } finally {
+ db.get = origGet;
+ process.off('unhandledRejection', onUnhandled);
+ }
+
+ console.log('✅ multipart part completion: a rejected getSession answers 500, no unhandled rejection and no hang');
+ process.exit(0);
+})().catch((err) => { console.error(err); process.exit(1); });
new file mode 100644
new file mode 100644
@@ -0,0 +1,50 @@
@@ -0,0 +1,50 @@
+'use strict';
+// generateFromVideo's promise had no reject path: a throw inside the async 'close' listener (here the
+// ffmpeg spawn) left it forever unsettled, so the awaiting request hung, and the rejection was
+// unhandled. The listener now catches and resolves null.
+const assert = require('assert');
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+const { EventEmitter } = require('events');
+
+(async () => {
+ const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-thumbthrow-'));
+ process.env.THUMBNAILS_PATH = path.join(tmp, 'thumbs');
+ process.env.VOD_PATH = path.join(tmp, 'vods');
+ process.env.FILES_PATH = path.join(tmp, 'files');
+
+ // The probe's ffprobe child closes cleanly; the ffmpeg spawn that follows throws.
+ const cp = require('child_process');
+ const realSpawn = cp.spawn;
+ cp.spawn = (cmd) => {
+ if (cmd !== 'ffprobe') throw new Error(`cannot spawn ${cmd}`);
+ const child = new EventEmitter();
+ child.stdout = new EventEmitter();
+ child.stderr = new EventEmitter();
+ setImmediate(() => child.emit('close', 0));
+ return child;
+ };
+ let unhandled = null;
+ const onUnhandled = (e) => { unhandled = e; };
+ process.on('unhandledRejection', onUnhandled);
+
+ const file = path.join(tmp, 'v.mp4');
+ fs.writeFileSync(file, 'not really a video');
+ try {
+ const thumb = require('../server/thumbnails/thumbnail-service');
+ // seekSeconds skips the probe-duration branch, so the close handler goes straight to ffmpeg.
+ const p = thumb.generateFromVideo(file, 'vod', 4242, { seekSeconds: 1 });
+ const result = await Promise.race([p, new Promise((r) => setTimeout(() => r('HUNG'), 3000))]);
+ await new Promise((r) => setTimeout(r, 50)); // let a stray unhandled rejection surface
+ assert.notStrictEqual(result, 'HUNG', 'the thumbnail request must settle when the close handler throws');
+ assert.strictEqual(result, null);
+ assert.strictEqual(unhandled, null, `unhandled rejection: ${unhandled && unhandled.message}`);
+ } finally {
+ cp.spawn = realSpawn;
+ process.off('unhandledRejection', onUnhandled);
+ }
+
+ console.log('✅ thumbnail probe: a throwing close handler resolves null instead of hanging');
+ process.exit(0);
+})().catch((err) => { console.error(err); process.exit(1); });