Zac Gaetano
91325a4267
DELETE /jobs/:id was throwing "404 not found" when the operator tried to
cancel a running job. BullMQ refuses job.remove() while a job is in the
active state; the route caught that error and fell through to the
404 branch, which was misleading because the job actually exists — the
queue was just refusing to drop it from under the worker.
Fix:
- Detect 'active' state explicitly and call moveToFailed(err, '0', false)
first. Token '0' bypasses the per-worker lock check (the operator-side
cancel doesn't hold the worker lock). That transitions active -> failed
and frees the queue's concurrency slot.
- If moveToFailed itself fails (lock owned by a live worker), fall back
to job.discard() so at least the result is thrown away.
- If remove() then fails (stalled, broken state), drop the job's Redis
key directly via queue.client. Last-resort obliteration.
- Stop swallowing getJob() errors — if Redis is sad, surface it via
next(err) instead of returning a misleading 404.
- Return { cancelled: true } when the job was active, so the client
can show "Cancelled" rather than "Removed" in any future toast.
While here: thumbnail jobs now run with concurrency 4 by default
(proxy 2, conform 1, import 1 unchanged). Every queue defaulted to
concurrency 1 before, so a single stalled job blocked the entire queue.
All three are overridable via PROXY_CONCURRENCY / THUMBNAIL_CONCURRENCY
/ CONFORM_CONCURRENCY env vars for nodes with more headroom.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
88 lines
3.3 KiB
JavaScript
88 lines
3.3 KiB
JavaScript
import 'dotenv/config';
|
|
import { Worker } from 'bullmq';
|
|
import { proxyWorker } from './workers/proxy.js';
|
|
import { thumbnailWorker } from './workers/thumbnail.js';
|
|
import { conformWorker } from './workers/conform.js';
|
|
import { youtubeImportWorker } from './workers/youtube-import.js';
|
|
import { startPromotionWorker } from './workers/promotion.js';
|
|
|
|
const parseRedisUrl = (url) => {
|
|
const parsed = new URL(url);
|
|
return {
|
|
host: parsed.hostname,
|
|
port: parseInt(parsed.port, 10),
|
|
password: parsed.password || undefined,
|
|
};
|
|
};
|
|
|
|
const redisOptions = parseRedisUrl(process.env.REDIS_URL || 'redis://localhost:6379');
|
|
|
|
const createWorker = (queueName, handler, overrides = {}) => {
|
|
const worker = new Worker(queueName, handler, {
|
|
connection: redisOptions,
|
|
// Stall detection: if a worker dies mid-job, BullMQ moves it back to wait
|
|
// after stalledInterval. Without this a crashed run sits in active forever.
|
|
stalledInterval: 30000,
|
|
maxStalledCount: 1,
|
|
lockDuration: 60000,
|
|
lockRenewTime: 15000,
|
|
...overrides,
|
|
});
|
|
|
|
worker.on('completed', (job) => {
|
|
console.log(`[${queueName}] Job ${job.id} completed`);
|
|
});
|
|
|
|
worker.on('failed', (job, err) => {
|
|
console.error(`[${queueName}] Job ${job.id} failed:`, err.message);
|
|
});
|
|
|
|
worker.on('stalled', (jobId) => {
|
|
console.warn(`[${queueName}] Job ${jobId} stalled — reclaimed`);
|
|
});
|
|
|
|
worker.on('progress', (job, progress) => {
|
|
console.log(`[${queueName}] Job ${job.id} progress:`, progress);
|
|
});
|
|
|
|
return worker;
|
|
};
|
|
|
|
// Per-queue concurrency. Defaults to 1, which serialises every job in a
|
|
// queue — meaning a single stalled job blocks every other one. We want
|
|
// thumbnails (cheap, parallel-safe) to run several at a time so a slow
|
|
// outlier doesn't back the rest of the catalog up. Proxy + conform are
|
|
// heavier (ffmpeg transcode) so we keep them lower to avoid trashing
|
|
// the box; tune via env if a node has more headroom.
|
|
const PROXY_CONCURRENCY = parseInt(process.env.PROXY_CONCURRENCY || '2', 10);
|
|
const THUMBNAIL_CONCURRENCY = parseInt(process.env.THUMBNAIL_CONCURRENCY || '4', 10);
|
|
const CONFORM_CONCURRENCY = parseInt(process.env.CONFORM_CONCURRENCY || '1', 10);
|
|
|
|
const workers = [
|
|
createWorker('proxy', proxyWorker, { concurrency: PROXY_CONCURRENCY }),
|
|
createWorker('thumbnail', thumbnailWorker, { concurrency: THUMBNAIL_CONCURRENCY }),
|
|
createWorker('conform', conformWorker, { concurrency: CONFORM_CONCURRENCY }),
|
|
// YouTube imports: keep concurrency at 1 so we don't burn through rate
|
|
// limits when several jobs land back-to-back. Lock window is longer than
|
|
// the default because a long video download can run for minutes.
|
|
createWorker('import', youtubeImportWorker, {
|
|
concurrency: 1,
|
|
lockDuration: 10 * 60 * 1000,
|
|
lockRenewTime: 60000,
|
|
}),
|
|
];
|
|
|
|
console.log(`Concurrency: proxy=${PROXY_CONCURRENCY} thumbnail=${THUMBNAIL_CONCURRENCY} conform=${CONFORM_CONCURRENCY} import=1`);
|
|
|
|
startPromotionWorker();
|
|
|
|
console.log('Wild Dragon Worker Service started');
|
|
console.log(`Redis: ${redisOptions.host}:${redisOptions.port}`);
|
|
console.log('Active queues: proxy, thumbnail, conform, import');
|
|
console.log('Background scans: promotion (growing-files → S3)');
|
|
|
|
process.on('SIGTERM', async () => {
|
|
console.log('SIGTERM received, shutting down...');
|
|
await Promise.all(workers.map(w => w.close()));
|
|
process.exit(0);
|
|
});
|