421 lines
16 KiB
JavaScript
421 lines
16 KiB
JavaScript
import { Worker } from 'worker_threads';
|
|
import { fileURLToPath } from 'url';
|
|
import os from 'os';
|
|
|
|
const FETCH_TIMEOUT_MS = 15000;
|
|
const MAX_BYTES = 50 * 1024 * 1024; // 50MB
|
|
|
|
// The crop/trim/resize/AVIF pipeline runs up to six sequential sharp passes (PNG
|
|
// intermediate, trim, extract, resize, AVIF encode). A sharp pipeline is owned by
|
|
// whichever thread created it, so running it on the main thread stalls every HTTP
|
|
// handler and every SSE client (heartbeats included). The pipeline therefore lives
|
|
// in imageCropWorker.js and is reached through the small pool below.
|
|
const WORKER_PATH = fileURLToPath(new URL('./imageCropWorker.js', import.meta.url));
|
|
const WORKER_BOOT_TIMEOUT_MS = 15000;
|
|
const DEFAULT_JOB_TIMEOUT_MS = 30000;
|
|
// libvips already saturates cores via its own per-process thread pool, so each worker
|
|
// is mostly IO/decode-bound rather than needing one thread per core. Cap at 3 to stay
|
|
// conservative on memory (each in-flight image can be tens of MB) while still covering
|
|
// the realistic burst. NOTE: use os.cpus(), not process.availableParallelism() — that
|
|
// helper is absent on some supported Node builds and silently collapsed this to 1.
|
|
const CPU_COUNT = Math.max(1, os.cpus?.()?.length || os.availableParallelism?.() || 2);
|
|
const POOL_SIZE = Math.max(1, Math.min(3, CPU_COUNT - 1));
|
|
|
|
/** @type {Array<{ worker: Worker, busy: boolean, id: number }>} */
|
|
const pool = [];
|
|
/** @type {Array<{ resolve: Function, reject: Function, timer: NodeJS.Timeout, inputBuffer: Buffer, timeoutMs: number }>} */
|
|
const queue = [];
|
|
let shuttingDown = false;
|
|
let workerSeq = 0;
|
|
/** Serializes pool growth so concurrent callers cannot overshoot POOL_SIZE. */
|
|
let growing = Promise.resolve();
|
|
|
|
/**
|
|
* Copies a Buffer into a standalone, transferable ArrayBuffer.
|
|
*
|
|
* `Buffer.from(buffer)` and `Uint8Array.prototype.subarray` can alias the shared
|
|
* 8 KB Buffer pool, and transferring pooled memory would detach it for every other
|
|
* Buffer in the process (including in-flight HTTP bodies). A fresh `Uint8Array` is
|
|
* always backed by its own ArrayBuffer, so it is always safe to transfer.
|
|
*
|
|
* @param {Buffer} buffer
|
|
* @returns {ArrayBuffer}
|
|
*/
|
|
function toTransferable(buffer) {
|
|
return Uint8Array.from(buffer).buffer;
|
|
}
|
|
|
|
/**
|
|
* Spawns a worker and resolves once it has booted far enough to accept a job.
|
|
* @returns {Promise<{ worker: Worker, busy: boolean, id: number }>}
|
|
*/
|
|
function spawnWorker() {
|
|
const entry = { worker: null, busy: false, id: ++workerSeq };
|
|
|
|
return new Promise((resolve, reject) => {
|
|
const bootTimer = setTimeout(() => {
|
|
cleanup();
|
|
try { entry.worker?.terminate(); } catch { /* already closed */ }
|
|
reject(new Error(`Image crop worker failed to start within ${WORKER_BOOT_TIMEOUT_MS} ms`));
|
|
}, WORKER_BOOT_TIMEOUT_MS);
|
|
|
|
function cleanup() {
|
|
clearTimeout(bootTimer);
|
|
entry.worker?.off('message', onBooted);
|
|
entry.worker?.off('error', onError);
|
|
entry.worker?.off('exit', onExit);
|
|
}
|
|
|
|
function onBooted(msg) {
|
|
// The worker sends an explicit `{ ready: true }` once its imports (notably
|
|
// sharp/libvips) have resolved. Without this handshake there is no first
|
|
// message to await, so a worker that fails to boot would only surface later
|
|
// as a confusing per-job failure.
|
|
if (!msg || msg.ready !== true) return;
|
|
cleanup();
|
|
resolve(entry);
|
|
}
|
|
|
|
function onError(err) {
|
|
cleanup();
|
|
reject(err instanceof Error ? err : new Error(err?.message || 'Image crop worker failed to start'));
|
|
}
|
|
|
|
function onExit(code) {
|
|
cleanup();
|
|
reject(new Error(`Image crop worker exited before becoming ready (code ${code})`));
|
|
}
|
|
|
|
try {
|
|
entry.worker = new Worker(WORKER_PATH);
|
|
} catch (err) {
|
|
cleanup();
|
|
reject(err);
|
|
return;
|
|
}
|
|
|
|
entry.worker.on('message', onBooted);
|
|
entry.worker.on('error', onError);
|
|
entry.worker.on('exit', onExit);
|
|
});
|
|
}
|
|
|
|
/**
|
|
* Returns an idle worker, spawning one when the pool is below capacity.
|
|
*
|
|
* When spawning fails the oldest queued job is failed with that error rather than
|
|
* being left waiting on a worker that will never arrive — otherwise a boot failure
|
|
* would deadlock the queue instead of surfacing to the HTTP route.
|
|
*
|
|
* @returns {Promise<{ worker: Worker, busy: boolean, id: number }>}
|
|
*/
|
|
async function acquireWorker() {
|
|
const idle = pool.find((entry) => !entry.busy);
|
|
if (idle) return idle;
|
|
|
|
if (pool.length < POOL_SIZE) {
|
|
// Serialize pool growth. Without this, N concurrent callers all observe
|
|
// `pool.length < POOL_SIZE` and each spawn a worker, overshooting the cap.
|
|
growing = growing.then(async () => {
|
|
if (pool.length >= POOL_SIZE) return;
|
|
try {
|
|
pool.push(await spawnWorker());
|
|
} catch (err) {
|
|
const waiting = queue[0];
|
|
if (!waiting) throw err;
|
|
// Fail the oldest queued job rather than leaving it waiting on a worker
|
|
// that will never arrive — otherwise a boot failure deadlocks the queue
|
|
// instead of surfacing to the HTTP route.
|
|
queue.shift();
|
|
clearTimeout(waiting.timer);
|
|
waiting.reject(err);
|
|
}
|
|
});
|
|
try {
|
|
await growing;
|
|
} catch {
|
|
// Already reported to the queued job; continue so this caller still queues.
|
|
}
|
|
// Re-check: an idle worker may now exist, or the pool may still be saturated.
|
|
if (pool.some((entry) => !entry.busy)) return;
|
|
}
|
|
|
|
// Every worker is busy and the pool is at capacity. enqueueJob/drainQueue hand the
|
|
// job the next free worker; acquiring one here is only needed to grow the pool.
|
|
return null;
|
|
}
|
|
|
|
/**
|
|
* Runs one job on a worker and resolves with its reply.
|
|
* @param {{ worker: Worker, busy: boolean, id: number }} entry
|
|
* @param {Buffer} inputBuffer
|
|
* @param {number} timeoutMs
|
|
*/
|
|
function runJob(entry, inputBuffer, timeoutMs) {
|
|
return new Promise((resolve, reject) => {
|
|
const jobId = `${entry.id}:${process.hrtime.bigint()}`;
|
|
const transfer = toTransferable(inputBuffer);
|
|
let settled = false;
|
|
|
|
const timer = setTimeout(() => {
|
|
// Terminate first, then report. Leaving a worker blocked in libvips would let
|
|
// it keep burning a core long after the HTTP request has been answered, and
|
|
// its eventual reply would then arrive against an already-settled job.
|
|
try { entry.worker.terminate(); } catch { /* already closed */ }
|
|
finish(new Error(`Image crop worker exceeded the ${timeoutMs} ms timeout and was terminated`));
|
|
}, timeoutMs);
|
|
|
|
function cleanup() {
|
|
clearTimeout(timer);
|
|
entry.worker.off('message', onMessage);
|
|
entry.worker.off('error', onError);
|
|
entry.worker.off('exit', onExit);
|
|
}
|
|
|
|
function finish(err, value) {
|
|
if (settled) return;
|
|
settled = true;
|
|
cleanup();
|
|
if (err) reject(err);
|
|
else resolve(value);
|
|
}
|
|
|
|
function onMessage(msg) {
|
|
// Ignore anything that is not this job's reply. Under `node --watch` the dev
|
|
// server's module-loader hook posts `watch:import` / `watch:require` notices on
|
|
// the worker's parentPort for every import, so the first message is not
|
|
// necessarily our reply. Job ids are matched instead of taking the first one.
|
|
if (!msg || msg.id !== jobId) return;
|
|
if (msg.ok) {
|
|
finish(null, {
|
|
buffer: Buffer.from(msg.buffer),
|
|
width: msg.width,
|
|
height: msg.height,
|
|
size: msg.size,
|
|
});
|
|
} else {
|
|
const err = new Error(msg.error || 'Image crop failed in worker');
|
|
if (msg.errorName) err.name = msg.errorName;
|
|
if (msg.stack) err.workerStack = msg.stack;
|
|
finish(err);
|
|
}
|
|
}
|
|
|
|
function onError(err) {
|
|
dropWorker(entry);
|
|
finish(err instanceof Error ? err : new Error(err?.message || 'Image crop worker error'));
|
|
}
|
|
|
|
function onExit(code) {
|
|
dropWorker(entry);
|
|
finish(new Error(`Image crop worker exited unexpectedly (code ${code})`));
|
|
}
|
|
|
|
entry.worker.on('message', onMessage);
|
|
entry.worker.on('error', onError);
|
|
entry.worker.on('exit', onExit);
|
|
|
|
try {
|
|
entry.worker.postMessage({ id: jobId, buffer: transfer }, [transfer]);
|
|
} catch (err) {
|
|
finish(err instanceof Error ? err : new Error(err?.message || 'Failed to post image to crop worker'));
|
|
}
|
|
});
|
|
}
|
|
|
|
/**
|
|
* Removes a dead worker from the pool. Jobs it was running have already been failed
|
|
* by its error/exit listener; queued jobs are re-homed by drainQueue().
|
|
* @param {{ worker: Worker }} entry
|
|
*/
|
|
function dropWorker(entry) {
|
|
const index = pool.indexOf(entry);
|
|
if (index !== -1) pool.splice(index, 1);
|
|
}
|
|
|
|
/**
|
|
* Starts as many queued jobs as there are idle workers.
|
|
*/
|
|
function drainQueue() {
|
|
while (queue.length > 0) {
|
|
const idle = pool.find((entry) => !entry.busy);
|
|
if (!idle) return;
|
|
|
|
const next = queue.shift();
|
|
clearTimeout(next.timer);
|
|
idle.busy = true;
|
|
|
|
runJob(idle, next.inputBuffer, next.timeoutMs).then(next.resolve, next.reject).finally(() => {
|
|
idle.busy = false;
|
|
drainQueue();
|
|
});
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Queues a crop job and resolves with the worker's result.
|
|
*
|
|
* `drainQueue` is the single owner of dispatch: it hands each queued job an idle
|
|
* worker and runs it, so a job can never be dispatched twice nor left waiting on a
|
|
* worker that acquireWorker never obtained.
|
|
*
|
|
* @param {Buffer} inputBuffer
|
|
* @param {number} timeoutMs
|
|
* @returns {Promise<{ buffer: Buffer, width: number, height: number, size: number }>}
|
|
*/
|
|
function enqueueJob(inputBuffer, timeoutMs) {
|
|
return new Promise((resolve, reject) => {
|
|
const timer = setTimeout(() => {
|
|
const index = queue.findIndex((item) => item.timer === timer);
|
|
if (index !== -1) queue.splice(index, 1);
|
|
reject(new Error(`Image crop worker queue timed out after ${timeoutMs} ms`));
|
|
}, timeoutMs);
|
|
|
|
queue.push({ resolve, reject, timer, inputBuffer, timeoutMs });
|
|
drainQueue();
|
|
});
|
|
}
|
|
|
|
/**
|
|
* Terminates every pooled worker and fails anything still queued.
|
|
*
|
|
* Call from shutdown so `node --watch` does not hang on "Waiting for graceful
|
|
* termination..." the way long-lived SSE sockets do.
|
|
*
|
|
* @returns {Promise<void>}
|
|
*/
|
|
export async function shutdownImageCropWorkers() {
|
|
shuttingDown = true;
|
|
|
|
while (queue.length > 0) {
|
|
const pending = queue.shift();
|
|
clearTimeout(pending.timer);
|
|
pending.reject(new Error('Image crop worker pool is shutting down'));
|
|
}
|
|
|
|
const entries = pool.splice(0, pool.length);
|
|
await Promise.all(
|
|
entries.map((entry) =>
|
|
Promise.resolve()
|
|
.then(() => entry.worker.terminate())
|
|
.catch(() => { /* already closed */ })
|
|
)
|
|
);
|
|
}
|
|
|
|
/**
|
|
* Validates and normalizes an image URL.
|
|
* @param {string} url
|
|
* @returns {URL}
|
|
*/
|
|
export function validateImageUrl(url) {
|
|
if (typeof url !== 'string' || !url.trim()) {
|
|
throw new Error('url is required');
|
|
}
|
|
let parsed;
|
|
try {
|
|
parsed = new URL(url.trim());
|
|
} catch {
|
|
throw new Error('Invalid URL');
|
|
}
|
|
if (parsed.protocol !== 'http:' && parsed.protocol !== 'https:') {
|
|
throw new Error(`Unsupported URL protocol: ${parsed.protocol}. Only http: and https: are allowed.`);
|
|
}
|
|
return parsed;
|
|
}
|
|
|
|
/**
|
|
* Crawls/downloads an image from a URL.
|
|
*
|
|
* @param {string} url
|
|
* @param {object} [options]
|
|
* @param {AbortSignal} [options.signal]
|
|
* @returns {Promise<Buffer>}
|
|
*/
|
|
export async function crawlImage(url, { signal } = {}) {
|
|
const parsedUrl = validateImageUrl(url);
|
|
|
|
const fetchSignal = signal
|
|
? AbortSignal.any([signal, AbortSignal.timeout(FETCH_TIMEOUT_MS)])
|
|
: AbortSignal.timeout(FETCH_TIMEOUT_MS);
|
|
|
|
const response = await fetch(parsedUrl.href, {
|
|
headers: {
|
|
'User-Agent':
|
|
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
|
|
'Accept': 'image/avif,image/webp,image/apng,image/svg+xml,image/*,*/*;q=0.8',
|
|
'Accept-Language': 'en-US,en;q=0.9,de;q=0.8',
|
|
'Referer': parsedUrl.origin + '/',
|
|
},
|
|
signal: fetchSignal,
|
|
redirect: 'follow',
|
|
});
|
|
|
|
if (!response.ok) {
|
|
throw new Error(`Failed to crawl image: HTTP ${response.status} ${response.statusText}`);
|
|
}
|
|
|
|
const contentLength = Number(response.headers.get('content-length'));
|
|
if (Number.isFinite(contentLength) && contentLength > MAX_BYTES) {
|
|
throw new Error(`Image size exceeds limit of ${MAX_BYTES} bytes`);
|
|
}
|
|
|
|
const arrayBuffer = await response.arrayBuffer();
|
|
const buffer = Buffer.from(arrayBuffer);
|
|
|
|
if (buffer.length === 0) {
|
|
throw new Error('Crawled image is empty (0 bytes)');
|
|
}
|
|
|
|
if (buffer.length > MAX_BYTES) {
|
|
throw new Error(`Image size (${buffer.length} bytes) exceeds limit of ${MAX_BYTES} bytes`);
|
|
}
|
|
|
|
return buffer;
|
|
}
|
|
|
|
/**
|
|
* Trims transparent and white borders from an image buffer, restores up to
|
|
* RESTORE_MARGIN px of the original image on each edge (never more than was
|
|
* actually trimmed, never enlarging beyond the original), resizes to max 1000px,
|
|
* and converts to AVIF.
|
|
*
|
|
* Runs the sharp pipeline in a pooled worker thread (see imageCropWorker.js) so the
|
|
* multi-pass encode never blocks the main event loop. Errors are re-thrown here with
|
|
* the worker's message intact, so callers that classify on message text (e.g. the
|
|
* `Failed to crawl image:` prefix check in /serper/image) keep working unchanged.
|
|
*
|
|
* @param {Buffer} inputBuffer - Raw image buffer
|
|
* @param {object} [options]
|
|
* @param {number} [options.timeout] - Hard wall-clock limit for the worker job
|
|
* @returns {Promise<{ buffer: Buffer, width: number, height: number, size: number }>}
|
|
*/
|
|
export async function cropAndConvertToAvif(inputBuffer, { timeout = DEFAULT_JOB_TIMEOUT_MS } = {}) {
|
|
if (!Buffer.isBuffer(inputBuffer)) {
|
|
throw new TypeError('cropAndConvertToAvif expects a Buffer');
|
|
}
|
|
if (shuttingDown) {
|
|
throw new Error('Image crop worker pool is shutting down');
|
|
}
|
|
|
|
// Grow the pool on demand, then queue. enqueueJob/drainQueue hand the job to the
|
|
// first idle worker, so an idle worker is never leased twice and a saturated pool
|
|
// simply waits instead of piling up sharp pipelines.
|
|
await acquireWorker();
|
|
return enqueueJob(inputBuffer, timeout);
|
|
}
|
|
|
|
/**
|
|
* Crawls a URL, crops white & transparency borders, resizes to max 1000px, and returns AVIF.
|
|
*
|
|
* @param {string} url - Image URL to crawl
|
|
* @param {object} [options]
|
|
* @param {AbortSignal} [options.signal] - Abort signal
|
|
* @param {number} [options.timeout] - Hard wall-clock limit for the crop worker
|
|
* @returns {Promise<{ buffer: Buffer, width: number, height: number, size: number }>}
|
|
*/
|
|
export async function crawlAndProcessImage(url, { signal, timeout } = {}) {
|
|
const rawBuffer = await crawlImage(url, { signal });
|
|
return cropAndConvertToAvif(rawBuffer, { timeout });
|
|
}
|