openvibers/openvibe.media Commits
Media: the per-tenant job cap counts one lane (paused transcodes no longer refuse thumbnails and hashes) (#103)
Changes
@@@ -671,7 +671,8 @@ that needs them GETs the job with its own tenant-scoped token (`GET /api/v2/:app**Idempotency.** `Idempotency-Key` (or `idempotency_key`) is unique per tenant: the same key with thesame request (type, object, params) answers with the job it made (`Idempotent-Replayed: true`); adifferent request under a used key is 409 `media.job.idempotency_conflict`. At most 50 jobs per tenant--may be queued or running (429 `media.job.too_many`).++may be queued or running **in each lane** (429 `media.job.too_many`): the heavy lane waits while a recording runs,++so a backlog of transcodes never stops a thumbnail, a content hash or a clip.**Cancellation, by the owner.** A proposed or queued job is cancelled at once. A running job gets`cancel_requested`; the worker aborts it (its ffmpeg is killed) and it ends `cancelled`. Finished@@@ -35,7 +35,10 @@ const events = require('../events');const STATUSES = ['proposed', 'queued', 'running', 'succeeded', 'failed', 'cancelled'];const FINISHED = ['succeeded', 'failed', 'cancelled'];const ACTIVE = ['queued', 'running'];--const MAX_ACTIVE_PER_TENANT = 50; // queued + running jobs one tenant may have (proposals excluded)++// Queued + running jobs one tenant may have IN ONE LANE (proposals excluded). Per lane: the heavy lane waits while a++// recording runs (worker.js), and on 2026-10-13 a day of back-to-back streams left 50 paused transcodes holding a single++// tenant-wide cap, so every thumbnail and content hash of the new recordings was refused (429) for hours.++const MAX_ACTIVE_PER_TENANT = 50;// Service-wide maintenance jobs (storage.orphans.scan) belong to no tenant: they run under this app id,// which has no apps row, so no API credential reaches them, and they announce no events.const SYSTEM_APP = '_media';@@@ -75,6 +78,11 @@ function typeSpec(type) {loadBuiltins();return TYPES.get(type) || null;}++/** The job types that run in `lane` (the per-tenant cap counts one lane). */++function typesInLane(lane) {++ loadBuiltins();++ return [...TYPES.values()].filter((s) => s.lane === lane).map((s) => s.type);++}function typeNames() {loadBuiltins();return [...TYPES.keys()];@@@ -266,9 +274,11 @@ async function enqueue({ appId, type, objectId = null, params = {}, status = 'quif (same) { out = { job: same, created: false, replayed: false, deduped: true }; return; }}if (status === 'queued') {-- const active = (await db.get("SELECT COUNT(*) AS n FROM media_jobs WHERE app_id = ? AND status IN ('queued', 'running')", [appId])).n;++ const lane = typesInLane(spec.lane);++ const active = (await db.get(`SELECT COUNT(*) AS n FROM media_jobs WHERE app_id = ? AND status IN ('queued', 'running')++ AND job_type IN (${lane.map(() => '?').join(', ')})`, [appId, ...lane])).n;if (active >= MAX_ACTIVE_PER_TENANT) {-- throw new JobError('media.job.too_many', `At most ${MAX_ACTIVE_PER_TENANT} jobs may be queued or running per tenant`, { status: 429, retryAfterS: 60 });++ throw new JobError('media.job.too_many', `At most ${MAX_ACTIVE_PER_TENANT} ${spec.lane} jobs may be queued or running per tenant`, { status: 429, retryAfterS: 60 });}}await db.run(`INSERT INTO media_jobs (id, app_id, object_id, job_type, status, idempotency_key, request_hash, params, max_attempts,@@@ -488,7 +498,7 @@ async function prune({ days = 30, types = ['thumbnail.regenerate', 'object.hash'module.exports = {STATUSES, FINISHED, ACTIVE, MAX_ACTIVE_PER_TENANT, SYSTEM_APP, JobError, bus,-- register, typeSpec, typeNames, requestHash, parseJson,++ register, typeSpec, typeNames, typesInLane, requestHash, parseJson,get, getForApp, jobPublic, jobEvent, list, counts,enqueue, approve, cancel,claim, renew, saveCheckpoint, succeed, fail, markCancelled, recoverInterrupted, waitFor, prune, staleStats,nnew file mode 100644@@@ -0,0 +1,56 @@++'use strict';++// The per-tenant job cap counts one lane. On 2026-10-13 the heavy lane waited out a day of back-to-back recordings++// (worker.js runs no transcode while a recording does) with 30 object.cmaf and 20 object.pack jobs queued: they held the++// whole tenant-wide cap of 50, so every thumbnail and content hash of the new recordings was refused with 429 for hours.++const assert = require('assert');++const fs = require('fs');++const os = require('os');++const path = require('path');++const http = require('http');++++const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-jobcap-'));++const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };++Object.assign(process.env, {++ VOD_PATH: dir('vods'), CLIPS_PATH: dir('clips'), FILES_PATH: dir('files'),++ THUMBNAILS_PATH: dir('thumbnails'), PASTES_PATH: dir('pastes'), OBJECTS_PATH: dir('objects'),++ MEDIA_INVARIANT_SCAN_HOURS: '0',++});++++(async () => {++ const stub = http.createServer((req, res) => {++ res.setHeader('Content-Type', 'application/json');++ if (req.url === '/oauth/token') return res.end(JSON.stringify({ access_token: 'tok', token_type: 'Bearer', expires_in: 300 }));++ res.statusCode = 404; res.end('{}');++ });++ await new Promise((r) => stub.listen(0, '127.0.0.1', r));++ const base = `http://127.0.0.1:${stub.address().port}`;++ const db = require('../server/db/database');++ await db.upsertApp({ app_id: 'live', api_key: 'live-key-jobcap' });++ await db.upsertApp({ app_id: 'games', api_key: 'games-key-jobcap' });++ require('../server/events').init({ eventsUrl: base, clientSecret: 's', networkUrl: base, intervalMs: 60000 });++ const queue = require('../server/jobs/queue');++ // Nothing runs these: no worker is started, so they stay queued like the paused heavy lane.++ queue.register('test.cap.heavy', { lane: 'heavy', run: async () => ({}) });++ queue.register('test.cap.light', { lane: 'light', run: async () => ({}) });++ const cap = queue.MAX_ACTIVE_PER_TENANT;++++ assert.ok(queue.typesInLane('heavy').includes('object.cmaf') && queue.typesInLane('heavy').includes('object.pack'), 'transcodes are heavy');++ assert.ok(queue.typesInLane('light').includes('thumbnail.regenerate') && !queue.typesInLane('light').includes('object.cmaf'), 'thumbnails are light');++++ for (let i = 0; i < cap; i++) await queue.enqueue({ appId: 'live', type: 'test.cap.heavy', params: { n: i } });++ await assert.rejects(queue.enqueue({ appId: 'live', type: 'test.cap.heavy', params: { n: cap } }),++ (e) => e.code === 'media.job.too_many' && e.status === 429 && /heavy jobs/.test(e.message), 'the heavy lane is full');++ const light = await queue.enqueue({ appId: 'live', type: 'test.cap.light', params: { n: 0 } });++ assert.ok(light.created, 'a full heavy lane does not stop a light job');++ assert.ok((await queue.enqueue({ appId: 'games', type: 'test.cap.heavy', params: { n: 0 } })).created, 'another tenant has its own lanes');++ // A proposal is not counted and is still accepted with a full lane.++ assert.ok((await queue.enqueue({ appId: 'live', type: 'test.cap.heavy', params: { n: 'p' }, status: 'proposed' })).created, 'a proposal is not capped');++++ for (let i = 1; i < cap; i++) await queue.enqueue({ appId: 'live', type: 'test.cap.light', params: { n: i } });++ await assert.rejects(queue.enqueue({ appId: 'live', type: 'test.cap.light', params: { n: cap } }),++ (e) => e.code === 'media.job.too_many' && /light jobs/.test(e.message), 'the light lane has the same cap of its own');++++ console.log('job-cap-per-lane: ok');++ stub.close();++ process.exit(0);++})().catch((err) => { console.error(err); process.exit(1); });
@@ -671,7 +671,8 @@ that needs them GETs the job with its own tenant-scoped token (`GET /api/v2/:app
@@ -671,7 +671,8 @@ that needs them GETs the job with its own tenant-scoped token (`GET /api/v2/:app
**Idempotency.** `Idempotency-Key` (or `idempotency_key`) is unique per tenant: the same key with the
**Idempotency.** `Idempotency-Key` (or `idempotency_key`) is unique per tenant: the same key with the
same request (type, object, params) answers with the job it made (`Idempotent-Replayed: true`); a
same request (type, object, params) answers with the job it made (`Idempotent-Replayed: true`); a
different request under a used key is 409 `media.job.idempotency_conflict`. At most 50 jobs per tenant
different request under a used key is 409 `media.job.idempotency_conflict`. At most 50 jobs per tenant
-may be queued or running (429 `media.job.too_many`).
+may be queued or running **in each lane** (429 `media.job.too_many`): the heavy lane waits while a recording runs,
+so a backlog of transcodes never stops a thumbnail, a content hash or a clip.
**Cancellation, by the owner.** A proposed or queued job is cancelled at once. A running job gets
**Cancellation, by the owner.** A proposed or queued job is cancelled at once. A running job gets
`cancel_requested`; the worker aborts it (its ffmpeg is killed) and it ends `cancelled`. Finished
`cancel_requested`; the worker aborts it (its ffmpeg is killed) and it ends `cancelled`. Finished
@@ -35,7 +35,10 @@ const events = require('../events');
@@ -35,7 +35,10 @@ const events = require('../events');
const STATUSES = ['proposed', 'queued', 'running', 'succeeded', 'failed', 'cancelled'];
const STATUSES = ['proposed', 'queued', 'running', 'succeeded', 'failed', 'cancelled'];
const FINISHED = ['succeeded', 'failed', 'cancelled'];
const FINISHED = ['succeeded', 'failed', 'cancelled'];
const ACTIVE = ['queued', 'running'];
const ACTIVE = ['queued', 'running'];
-const MAX_ACTIVE_PER_TENANT = 50; // queued + running jobs one tenant may have (proposals excluded)
+// Queued + running jobs one tenant may have IN ONE LANE (proposals excluded). Per lane: the heavy lane waits while a
+// recording runs (worker.js), and on 2026-10-13 a day of back-to-back streams left 50 paused transcodes holding a single
+// tenant-wide cap, so every thumbnail and content hash of the new recordings was refused (429) for hours.
+const MAX_ACTIVE_PER_TENANT = 50;
// Service-wide maintenance jobs (storage.orphans.scan) belong to no tenant: they run under this app id,
// Service-wide maintenance jobs (storage.orphans.scan) belong to no tenant: they run under this app id,
// which has no apps row, so no API credential reaches them, and they announce no events.
// which has no apps row, so no API credential reaches them, and they announce no events.
const SYSTEM_APP = '_media';
const SYSTEM_APP = '_media';
@@ -75,6 +78,11 @@ function typeSpec(type) {
@@ -75,6 +78,11 @@ function typeSpec(type) {
loadBuiltins();
loadBuiltins();
return TYPES.get(type) || null;
return TYPES.get(type) || null;
}
}
+/** The job types that run in `lane` (the per-tenant cap counts one lane). */
+function typesInLane(lane) {
+ loadBuiltins();
+ return [...TYPES.values()].filter((s) => s.lane === lane).map((s) => s.type);
+}
function typeNames() {
function typeNames() {
loadBuiltins();
loadBuiltins();
return [...TYPES.keys()];
return [...TYPES.keys()];
@@ -266,9 +274,11 @@ async function enqueue({ appId, type, objectId = null, params = {}, status = 'qu
@@ -266,9 +274,11 @@ async function enqueue({ appId, type, objectId = null, params = {}, status = 'qu
if (same) { out = { job: same, created: false, replayed: false, deduped: true }; return; }
if (same) { out = { job: same, created: false, replayed: false, deduped: true }; return; }
}
}
if (status === 'queued') {
if (status === 'queued') {
- const active = (await db.get("SELECT COUNT(*) AS n FROM media_jobs WHERE app_id = ? AND status IN ('queued', 'running')", [appId])).n;
+ const lane = typesInLane(spec.lane);
+ const active = (await db.get(`SELECT COUNT(*) AS n FROM media_jobs WHERE app_id = ? AND status IN ('queued', 'running')
+ AND job_type IN (${lane.map(() => '?').join(', ')})`, [appId, ...lane])).n;
if (active >= MAX_ACTIVE_PER_TENANT) {
if (active >= MAX_ACTIVE_PER_TENANT) {
- throw new JobError('media.job.too_many', `At most ${MAX_ACTIVE_PER_TENANT} jobs may be queued or running per tenant`, { status: 429, retryAfterS: 60 });
+ throw new JobError('media.job.too_many', `At most ${MAX_ACTIVE_PER_TENANT} ${spec.lane} jobs may be queued or running per tenant`, { status: 429, retryAfterS: 60 });
}
}
}
}
await db.run(`INSERT INTO media_jobs (id, app_id, object_id, job_type, status, idempotency_key, request_hash, params, max_attempts,
await db.run(`INSERT INTO media_jobs (id, app_id, object_id, job_type, status, idempotency_key, request_hash, params, max_attempts,
@@ -488,7 +498,7 @@ async function prune({ days = 30, types = ['thumbnail.regenerate', 'object.hash'
@@ -488,7 +498,7 @@ async function prune({ days = 30, types = ['thumbnail.regenerate', 'object.hash'
module.exports = {
module.exports = {
STATUSES, FINISHED, ACTIVE, MAX_ACTIVE_PER_TENANT, SYSTEM_APP, JobError, bus,
STATUSES, FINISHED, ACTIVE, MAX_ACTIVE_PER_TENANT, SYSTEM_APP, JobError, bus,
- register, typeSpec, typeNames, requestHash, parseJson,
+ register, typeSpec, typeNames, typesInLane, requestHash, parseJson,
get, getForApp, jobPublic, jobEvent, list, counts,
get, getForApp, jobPublic, jobEvent, list, counts,
enqueue, approve, cancel,
enqueue, approve, cancel,
claim, renew, saveCheckpoint, succeed, fail, markCancelled, recoverInterrupted, waitFor, prune, staleStats,
claim, renew, saveCheckpoint, succeed, fail, markCancelled, recoverInterrupted, waitFor, prune, staleStats,
new file mode 100644
new file mode 100644
@@ -0,0 +1,56 @@
@@ -0,0 +1,56 @@
+'use strict';
+// The per-tenant job cap counts one lane. On 2026-10-13 the heavy lane waited out a day of back-to-back recordings
+// (worker.js runs no transcode while a recording does) with 30 object.cmaf and 20 object.pack jobs queued: they held the
+// whole tenant-wide cap of 50, so every thumbnail and content hash of the new recordings was refused with 429 for hours.
+const assert = require('assert');
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+const http = require('http');
+
+const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-jobcap-'));
+const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };
+Object.assign(process.env, {
+ VOD_PATH: dir('vods'), CLIPS_PATH: dir('clips'), FILES_PATH: dir('files'),
+ THUMBNAILS_PATH: dir('thumbnails'), PASTES_PATH: dir('pastes'), OBJECTS_PATH: dir('objects'),
+ MEDIA_INVARIANT_SCAN_HOURS: '0',
+});
+
+(async () => {
+ const stub = http.createServer((req, res) => {
+ res.setHeader('Content-Type', 'application/json');
+ if (req.url === '/oauth/token') return res.end(JSON.stringify({ access_token: 'tok', token_type: 'Bearer', expires_in: 300 }));
+ res.statusCode = 404; res.end('{}');
+ });
+ await new Promise((r) => stub.listen(0, '127.0.0.1', r));
+ const base = `http://127.0.0.1:${stub.address().port}`;
+ const db = require('../server/db/database');
+ await db.upsertApp({ app_id: 'live', api_key: 'live-key-jobcap' });
+ await db.upsertApp({ app_id: 'games', api_key: 'games-key-jobcap' });
+ require('../server/events').init({ eventsUrl: base, clientSecret: 's', networkUrl: base, intervalMs: 60000 });
+ const queue = require('../server/jobs/queue');
+ // Nothing runs these: no worker is started, so they stay queued like the paused heavy lane.
+ queue.register('test.cap.heavy', { lane: 'heavy', run: async () => ({}) });
+ queue.register('test.cap.light', { lane: 'light', run: async () => ({}) });
+ const cap = queue.MAX_ACTIVE_PER_TENANT;
+
+ assert.ok(queue.typesInLane('heavy').includes('object.cmaf') && queue.typesInLane('heavy').includes('object.pack'), 'transcodes are heavy');
+ assert.ok(queue.typesInLane('light').includes('thumbnail.regenerate') && !queue.typesInLane('light').includes('object.cmaf'), 'thumbnails are light');
+
+ for (let i = 0; i < cap; i++) await queue.enqueue({ appId: 'live', type: 'test.cap.heavy', params: { n: i } });
+ await assert.rejects(queue.enqueue({ appId: 'live', type: 'test.cap.heavy', params: { n: cap } }),
+ (e) => e.code === 'media.job.too_many' && e.status === 429 && /heavy jobs/.test(e.message), 'the heavy lane is full');
+ const light = await queue.enqueue({ appId: 'live', type: 'test.cap.light', params: { n: 0 } });
+ assert.ok(light.created, 'a full heavy lane does not stop a light job');
+ assert.ok((await queue.enqueue({ appId: 'games', type: 'test.cap.heavy', params: { n: 0 } })).created, 'another tenant has its own lanes');
+ // A proposal is not counted and is still accepted with a full lane.
+ assert.ok((await queue.enqueue({ appId: 'live', type: 'test.cap.heavy', params: { n: 'p' }, status: 'proposed' })).created, 'a proposal is not capped');
+
+ for (let i = 1; i < cap; i++) await queue.enqueue({ appId: 'live', type: 'test.cap.light', params: { n: i } });
+ await assert.rejects(queue.enqueue({ appId: 'live', type: 'test.cap.light', params: { n: cap } }),
+ (e) => e.code === 'media.job.too_many' && /light jobs/.test(e.message), 'the light lane has the same cap of its own');
+
+ console.log('job-cap-per-lane: ok');
+ stub.close();
+ process.exit(0);
+})().catch((err) => { console.error(err); process.exit(1); });