80 lines
3.8 KiB
JavaScript
80 lines
3.8 KiB
JavaScript
import dbHandler, { holdEventLoop, resolveClient } from "../db-handler";
|
|
import { quoteIdentifier } from "../../utils/sql-identifier";
|
|
import { QUEUE_NOW_SQL, requireQueueTable } from "./queue-utils";
|
|
/**
|
|
* Claim the next available jobs, oldest `queue_available_at` first. Each
|
|
* returned row carries a `queue_lock_token` that must be passed to `ack`,
|
|
* `fail` or `extend`.
|
|
*
|
|
* `FOR UPDATE SKIP LOCKED` lets concurrent workers claim different jobs
|
|
* without blocking on each other or claiming the same job twice.
|
|
*/
|
|
export default async function QueueClaim({ table, limit = 1, lease_ms, max_attempts, config, }) {
|
|
try {
|
|
const options = requireQueueTable(table);
|
|
const tableSql = quoteIdentifier(table);
|
|
const finalLimit = Math.max(1, Math.floor(Number(limit) || 1));
|
|
const leaseMs = Math.max(0, Math.floor(Number(lease_ms ?? options.lease_ms) || 0));
|
|
const maxAttempts = Math.max(1, Math.floor(Number(max_attempts ?? options.max_attempts) || 1));
|
|
/**
|
|
* Release jobs whose worker died or stalled past its lease: back to
|
|
* `pending`, or `failed` if that was the last attempt.
|
|
*/
|
|
const expired_res = await dbHandler({
|
|
query: `UPDATE ${tableSql} SET queue_status = IF(queue_attempts >= ?, 'failed', 'pending'), queue_lease_expires_at = NULL, queue_lock_token = NULL, queue_last_error = 'Lease expired', updated_at = ${QUEUE_NOW_SQL} WHERE queue_status = 'processing' AND queue_lease_expires_at <= ${QUEUE_NOW_SQL}`,
|
|
values: [maxAttempts],
|
|
config,
|
|
});
|
|
if (!expired_res.success) {
|
|
return { ...expired_res, success: false };
|
|
}
|
|
const CLIENT = resolveClient(config);
|
|
if (!CLIENT) {
|
|
throw new Error(`Couldn't grab MariaDB Client.`);
|
|
}
|
|
const lock_token = crypto.randomUUID();
|
|
return await holdEventLoop(() => CLIENT.begin(async (tx) => {
|
|
/**
|
|
* The ORDER BY must match `idx_queue_claim`. A filesort would
|
|
* lock every pending row before LIMIT applies, and SKIP LOCKED
|
|
* would then hide the whole queue from other workers.
|
|
*/
|
|
const matched_res = await dbHandler({
|
|
query: `SELECT id FROM ${tableSql} FORCE INDEX (idx_queue_claim) WHERE queue_status = 'pending' AND queue_available_at <= ${QUEUE_NOW_SQL} ORDER BY queue_status, queue_available_at, id LIMIT ? FOR UPDATE SKIP LOCKED`,
|
|
values: [finalLimit],
|
|
client: tx,
|
|
});
|
|
if (!matched_res.success) {
|
|
return {
|
|
...matched_res,
|
|
success: false,
|
|
};
|
|
}
|
|
const ids = (matched_res.payload || []).map((row) => row.id);
|
|
if (ids.length === 0) {
|
|
return { success: true, payload: [] };
|
|
}
|
|
const idPlaceholders = ids.map(() => "?").join(",");
|
|
const update_res = await dbHandler({
|
|
query: `UPDATE ${tableSql} SET queue_status = 'processing', queue_attempts = queue_attempts + 1, queue_lease_expires_at = ${QUEUE_NOW_SQL} + ?, queue_lock_token = ?, updated_at = ${QUEUE_NOW_SQL} WHERE id IN (${idPlaceholders})`,
|
|
values: [leaseMs, lock_token, ...ids],
|
|
client: tx,
|
|
});
|
|
if (!update_res.success) {
|
|
throw new Error(update_res.error || "Queue claim failed");
|
|
}
|
|
return await dbHandler({
|
|
query: `SELECT * FROM ${tableSql} WHERE id IN (${idPlaceholders}) ORDER BY queue_available_at, id`,
|
|
values: ids,
|
|
client: tx,
|
|
});
|
|
}));
|
|
}
|
|
catch (error) {
|
|
return {
|
|
success: false,
|
|
error: error.message,
|
|
};
|
|
}
|
|
}
|