seb/dashback
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

u

Browse Source
This commit is contained in:
seb
2026-10-04 23:39:49 +02:00
parent 9feca54a0b
commit ab40a38a75
10 changed files with 1217 additions and 865 deletions
Download Patch File Download Diff File Expand all files Collapse all files

402
src/services/imageCrop.js
View File

@@ -1,12 +1,302 @@
import sharp from 'sharp';
import { Worker } from 'worker_threads';
import { fileURLToPath } from 'url';
const FETCH_TIMEOUT_MS = 15000;
const MAX_BYTES = 50 * 1024 * 1024; // 50MB
const MAX_EDGE = 1000;
const TRIM_THRESHOLD = 10;
const RESTORE_MARGIN = 5; // px restored per edge after trimming (up to what was trimmed)
const AVIF_QUALITY = 80;
const AVIF_EFFORT = 4;
// 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 keeps its own warm thread pool per process, so one worker per CPU is
// plenty; sharp's internal concurrency is what actually saturates a worker.
const POOL_SIZE = Math.max(1, Math.min(4, (process.availableParallelism?.() ?? 2) - 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.
@@ -85,90 +375,29 @@ export async function crawlImage(url, { signal } = {}) {
* 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) {
// 1. Initial metadata and orientation correction
const baseSharp = sharp(inputBuffer, { failOn: 'none' }).rotate();
const initialMeta = await baseSharp.metadata();
const hasAlpha = Boolean(initialMeta.hasAlpha);
// Use PNG intermediate for trimming passes to preserve alpha
const originalBuffer = await baseSharp.png().toBuffer();
const origMeta = await sharp(originalBuffer).metadata();
// Cumulative crop box in original (oriented) pixel coordinates
let box = { left: 0, top: 0, width: origMeta.width, height: origMeta.height };
let currentBuffer = originalBuffer;
const trimPass = async (background) => {
try {
const trimmed = await sharp(currentBuffer)
.trim({ background, threshold: TRIM_THRESHOLD })
.png()
.toBuffer({ resolveWithObject: true });
if (trimmed.info.width > 0 && trimmed.info.height > 0) {
currentBuffer = trimmed.data;
box = {
left: box.left - (trimmed.info.trimOffsetLeft ?? 0),
top: box.top - (trimmed.info.trimOffsetTop ?? 0),
width: trimmed.info.width,
height: trimmed.info.height,
};
}
} catch {
// Keep currentBuffer
}
};
// 2. Step A: Trim transparent borders if alpha is present
if (hasAlpha) await trimPass({ r: 0, g: 0, b: 0, alpha: 0 });
// 3. Step B: Trim white borders
await trimPass('#ffffff');
// 4. Step C: Trim transparent borders again (handles white border containing transparent padding)
if (hasAlpha) await trimPass({ r: 0, g: 0, b: 0, alpha: 0 });
// 4b. Restore up to RESTORE_MARGIN px per edge from the original, clamped to original bounds
const left = Math.max(0, box.left - RESTORE_MARGIN);
const top = Math.max(0, box.top - RESTORE_MARGIN);
const right = Math.min(origMeta.width, box.left + box.width + RESTORE_MARGIN);
const bottom = Math.min(origMeta.height, box.top + box.height + RESTORE_MARGIN);
if (left !== box.left || top !== box.top || right - left !== box.width || bottom - top !== box.height) {
try {
const restored = await sharp(originalBuffer)
.extract({ left, top, width: right - left, height: bottom - top })
.png()
.toBuffer({ resolveWithObject: true });
currentBuffer = restored.data;
} catch {
// Keep currentBuffer
}
export async function cropAndConvertToAvif(inputBuffer, { timeout = DEFAULT_JOB_TIMEOUT_MS } = {}) {
if (!Buffer.isBuffer(inputBuffer)) {
throw new TypeError('cropAndConvertToAvif expects a Buffer');
}
const currentMeta = await sharp(currentBuffer).metadata();
// 5. Resize to max 1000px without enlargement
let finalPipeline = sharp(currentBuffer);
if (currentMeta.width > MAX_EDGE || currentMeta.height > MAX_EDGE) {
finalPipeline = finalPipeline.resize(MAX_EDGE, MAX_EDGE, {
fit: 'inside',
withoutEnlargement: true,
});
if (shuttingDown) {
throw new Error('Image crop worker pool is shutting down');
}
// 6. Convert to AVIF
const avifResult = await finalPipeline
.avif({ quality: AVIF_QUALITY, effort: AVIF_EFFORT })
.toBuffer({ resolveWithObject: true });
return {
buffer: avifResult.data,
width: avifResult.info.width,
height: avifResult.info.height,
size: avifResult.info.size,
};
// 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);
}
/**
@@ -177,9 +406,10 @@ export async function cropAndConvertToAvif(inputBuffer) {
* @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 } = {}) {
export async function crawlAndProcessImage(url, { signal, timeout } = {}) {
const rawBuffer = await crawlImage(url, { signal });
return cropAndConvertToAvif(rawBuffer);
return cropAndConvertToAvif(rawBuffer, { timeout });
}