diff --git a/dist/functions/grab-mariadb-client.js b/dist/functions/grab-mariadb-client.js index 8f73a1f..7fcc723 100644 --- a/dist/functions/grab-mariadb-client.js +++ b/dist/functions/grab-mariadb-client.js @@ -4,6 +4,8 @@ import {} from "../types"; import { SQL } from "bun"; /** Reuse one client per distinct connection config to avoid exhausting MariaDB connections. */ const clientCache = new Map(); +/** Cache key each client was created with */ +const clientKeys = new WeakMap(); /** * Cache key covering everything that affects the connection, not just the * database name. @@ -24,8 +26,7 @@ export default function grabMariaDBClient({ config, }) { } // Prefer the process-wide client when it was created from the same config if (global.MARIADB_CLIENT && - global.CONFIG && - mariadbClientCacheKey(global.CONFIG) === cacheKey) { + clientKeys.get(global.MARIADB_CLIENT) === cacheKey) { clientCache.set(cacheKey, global.MARIADB_CLIENT); return global.MARIADB_CLIENT; } @@ -53,6 +54,21 @@ export default function grabMariaDBClient({ config, }) { ...config.db_config, adapter: "mariadb", }); + /** + * A closed client can't run queries, so forget it on close: the next + * grab (or `ensureInit`) creates a fresh one. + */ + const close = MariaDBClient.close.bind(MariaDBClient); + MariaDBClient.close = (...args) => { + if (clientCache.get(cacheKey) === MariaDBClient) { + clientCache.delete(cacheKey); + } + if (global.MARIADB_CLIENT === MariaDBClient) { + global.MARIADB_CLIENT = undefined; + } + return close(...args); + }; + clientKeys.set(MariaDBClient, cacheKey); clientCache.set(cacheKey, MariaDBClient); return MariaDBClient; } diff --git a/dist/functions/init.d.ts b/dist/functions/init.d.ts index 3ee4554..35ef673 100644 --- a/dist/functions/init.d.ts +++ b/dist/functions/init.d.ts @@ -5,7 +5,8 @@ import { type BunMariaDBConfig, type BUN_MARIADB_DatabaseSchemaType } from "../t declare global { var CONFIG: BunMariaDBConfig; var DB_SCHEMA: BUN_MARIADB_DatabaseSchemaType; - var MARIADB_CLIENT: Bun.SQL; + /** Unset after the client is closed */ + var MARIADB_CLIENT: Bun.SQL | undefined; var CONFIG_FILE_PATH: string; var SCHEMA_FILE_PATH: string; } diff --git a/dist/index.d.ts b/dist/index.d.ts index bb37a83..66d0e91 100644 --- a/dist/index.d.ts +++ b/dist/index.d.ts @@ -3,6 +3,12 @@ import DbInsert from "./lib/mariadb/db-insert"; import DbSelect from "./lib/mariadb/db-select"; import DbSQL from "./lib/mariadb/db-sql"; import DbUpdate from "./lib/mariadb/db-update"; +import QueueAck from "./lib/queue/queue-ack"; +import QueueCancel from "./lib/queue/queue-cancel"; +import QueueClaim from "./lib/queue/queue-claim"; +import QueueExtend from "./lib/queue/queue-extend"; +import QueueFail from "./lib/queue/queue-fail"; +import QueuePush from "./lib/queue/queue-push"; import grabDbSchema from "./utils/grab-db-schema"; import grabJoinFieldsFromQueryObject from "./utils/grab-join-fields-from-query-object"; declare const BunMariaDB: { @@ -11,6 +17,14 @@ declare const BunMariaDB: { readonly update: typeof DbUpdate; readonly delete: typeof DbDelete; readonly sql: typeof DbSQL; + readonly queue: { + readonly push: typeof QueuePush; + readonly claim: typeof QueueClaim; + readonly ack: typeof QueueAck; + readonly fail: typeof QueueFail; + readonly extend: typeof QueueExtend; + readonly cancel: typeof QueueCancel; + }; readonly utils: { readonly grab_db_schema: typeof grabDbSchema; readonly grab_join_fields_from_query_object: typeof grabJoinFieldsFromQueryObject; @@ -18,4 +32,4 @@ declare const BunMariaDB: { }; export default BunMariaDB; export type * from "./types"; -export { UsersOmitedFields, MariaDBCollations, MariaDBCharsets, TextFieldTypesArray, BUN_MARIADB_DATATYPES, MariaDBIndexTypes, ServerQueryOperators, ServerQueryEqualities, SQlComparisons, DataCrudRequestMethods, DataCrudRequestMethodsLowerCase, DsqlCrudActions, QueryFields, DockerComposeServices, IndexTypes, DefaultFields, RequiredENVs, } from "./types"; +export { UsersOmitedFields, MariaDBCollations, MariaDBCharsets, TextFieldTypesArray, BUN_MARIADB_DATATYPES, MariaDBIndexTypes, ServerQueryOperators, ServerQueryEqualities, SQlComparisons, DataCrudRequestMethods, DataCrudRequestMethodsLowerCase, DsqlCrudActions, QueryFields, DockerComposeServices, IndexTypes, DefaultFields, QueueStatuses, QueueFields, QueueProtectedFields, RequiredENVs, } from "./types"; diff --git a/dist/index.js b/dist/index.js index f2bb14f..2f95be6 100644 --- a/dist/index.js +++ b/dist/index.js @@ -4,6 +4,12 @@ import DbInsert from "./lib/mariadb/db-insert"; import DbSelect from "./lib/mariadb/db-select"; import DbSQL from "./lib/mariadb/db-sql"; import DbUpdate from "./lib/mariadb/db-update"; +import QueueAck from "./lib/queue/queue-ack"; +import QueueCancel from "./lib/queue/queue-cancel"; +import QueueClaim from "./lib/queue/queue-claim"; +import QueueExtend from "./lib/queue/queue-extend"; +import QueueFail from "./lib/queue/queue-fail"; +import QueuePush from "./lib/queue/queue-push"; import grabDbSchema from "./utils/grab-db-schema"; import grabJoinFieldsFromQueryObject from "./utils/grab-join-fields-from-query-object"; /** @@ -22,10 +28,18 @@ const BunMariaDB = { update: withInit(DbUpdate), delete: withInit(DbDelete), sql: withInit(DbSQL), + queue: { + push: withInit(QueuePush), + claim: withInit(QueueClaim), + ack: withInit(QueueAck), + fail: withInit(QueueFail), + extend: withInit(QueueExtend), + cancel: withInit(QueueCancel), + }, utils: { grab_db_schema: withInit(grabDbSchema), grab_join_fields_from_query_object: grabJoinFieldsFromQueryObject, }, }; export default BunMariaDB; -export { UsersOmitedFields, MariaDBCollations, MariaDBCharsets, TextFieldTypesArray, BUN_MARIADB_DATATYPES, MariaDBIndexTypes, ServerQueryOperators, ServerQueryEqualities, SQlComparisons, DataCrudRequestMethods, DataCrudRequestMethodsLowerCase, DsqlCrudActions, QueryFields, DockerComposeServices, IndexTypes, DefaultFields, RequiredENVs, } from "./types"; +export { UsersOmitedFields, MariaDBCollations, MariaDBCharsets, TextFieldTypesArray, BUN_MARIADB_DATATYPES, MariaDBIndexTypes, ServerQueryOperators, ServerQueryEqualities, SQlComparisons, DataCrudRequestMethods, DataCrudRequestMethodsLowerCase, DsqlCrudActions, QueryFields, DockerComposeServices, IndexTypes, DefaultFields, QueueStatuses, QueueFields, QueueProtectedFields, RequiredENVs, } from "./types"; diff --git a/dist/lib/db-handler.js b/dist/lib/db-handler.js index 06cc618..ca8068a 100644 --- a/dist/lib/db-handler.js +++ b/dist/lib/db-handler.js @@ -1,4 +1,4 @@ -import grabMariaDBClient, { mariadbClientCacheKey, } from "../functions/grab-mariadb-client"; +import grabMariaDBClient from "../functions/grab-mariadb-client"; /** * Resolve a shared MariaDB client. Prefer the global client so schema sync * and CRUD do not open a new connection per query. @@ -7,13 +7,7 @@ export function resolveClient(config) { if (!config) { return global.MARIADB_CLIENT; } - // Same connection config as the global client → reuse it - if (global.MARIADB_CLIENT && - (!global.CONFIG || - mariadbClientCacheKey(config) === - mariadbClientCacheKey(global.CONFIG))) { - return global.MARIADB_CLIENT; - } + // Returns the global client when it was created from the same config return grabMariaDBClient({ config }); } /** diff --git a/dist/lib/mariadb/db-delete.js b/dist/lib/mariadb/db-delete.js index d44f137..1482370 100644 --- a/dist/lib/mariadb/db-delete.js +++ b/dist/lib/mariadb/db-delete.js @@ -2,6 +2,7 @@ import dbHandler from "../db-handler"; import _ from "lodash"; import sqlGenerator from "../../utils/sql-generator"; import { quoteIdentifier } from "../../utils/sql-identifier"; +import { isQueueTable, QUEUE_NOT_CLAIMED_SQL } from "../queue/queue-utils"; export default async function DbDelete({ table, query, targetId, config, }) { let sqlObj = null; try { @@ -11,7 +12,7 @@ export default async function DbDelete({ table, query, targetId, config, }) { genObject: finalQuery, targetId, }); - const whereClause = sqlObj.where; + let whereClause = sqlObj.where; if (!whereClause) { return { success: false, @@ -21,6 +22,9 @@ export default async function DbDelete({ table, query, targetId, config, }) { }, }; } + if (isQueueTable(table)) { + whereClause += ` AND ${QUEUE_NOT_CLAIMED_SQL}`; + } let sql = `DELETE FROM ${quoteIdentifier(table)} ${whereClause}`; sql += ` RETURNING *`; sqlObj.string = sql; diff --git a/dist/lib/mariadb/db-insert.js b/dist/lib/mariadb/db-insert.js index 84851b4..5ca9529 100644 --- a/dist/lib/mariadb/db-insert.js +++ b/dist/lib/mariadb/db-insert.js @@ -2,9 +2,13 @@ import dbHandler from "../db-handler"; import sqlInsertGenerator from "../../utils/sql-insert-generator"; import { sanitizeHtmlFieldsBatch } from "../../utils/sanitize-html-fields"; import grabDuplicateSafeInsertSql from "../grab-duplicate-safe-insert-sql"; +import { assertNoProtectedQueueFields, isQueueTable, } from "../queue/queue-utils"; export default async function DbInsert({ table, data, update_on_duplicate, config, }) { let sqlObj = null; try { + if (isQueueTable(table)) { + assertNoProtectedQueueFields(table, data); + } const sanitizedData = sanitizeHtmlFieldsBatch({ table, data, diff --git a/dist/lib/mariadb/db-update.js b/dist/lib/mariadb/db-update.js index 9a64229..19770ec 100644 --- a/dist/lib/mariadb/db-update.js +++ b/dist/lib/mariadb/db-update.js @@ -4,6 +4,7 @@ import sanitizeHtmlFields from "../../utils/sanitize-html-fields"; import { quoteIdentifier } from "../../utils/sql-identifier"; import sqlValuePlaceholder from "../../utils/sql-value-placeholder"; import grabTableFields from "../../utils/grab-table-fields"; +import { assertNoProtectedQueueFields, isQueueTable, QUEUE_NOT_CLAIMED_SQL, } from "../queue/queue-utils"; /** Thrown inside the transaction to roll back and return a response */ class UpdateAbort extends Error { response; @@ -26,13 +27,17 @@ export default async function DbUpdate({ table, data, query, targetId, config, } genObject: finalQuery, targetId, }); - const whereClause = sqlQueryObj.where; + let whereClause = sqlQueryObj.where; if (!whereClause) { return { success: false, msg: `No WHERE clause`, }; } + if (isQueueTable(table)) { + assertNoProtectedQueueFields(table, [data]); + whereClause += ` AND ${QUEUE_NOT_CLAIMED_SQL}`; + } const sanitizedData = sanitizeHtmlFields({ table, data, diff --git a/dist/lib/queue/queue-ack.d.ts b/dist/lib/queue/queue-ack.d.ts new file mode 100644 index 0000000..6c751e2 --- /dev/null +++ b/dist/lib/queue/queue-ack.d.ts @@ -0,0 +1,17 @@ +import type { BunMariaDBConfig, DBResponseObject } from "../../types"; +type Params = { + table: Table; + id: number | string; + /** `queue_lock_token` from the claimed row */ + lock_token: string; + config?: BunMariaDBConfig; +}; +/** + * Mark a claimed job as done: deletes it, or sets it to `done` when the + * table has `keep_completed`. + * + * Fails when the lock token no longer matches, i.e. the lease expired and + * another worker reclaimed the job. + */ +export default function QueueAck
({ table, id, lock_token, config, }: Params
): Promise; +export {}; diff --git a/dist/lib/queue/queue-ack.js b/dist/lib/queue/queue-ack.js new file mode 100644 index 0000000..8de93f2 --- /dev/null +++ b/dist/lib/queue/queue-ack.js @@ -0,0 +1,61 @@ +import dbHandler from "../db-handler"; +import { quoteIdentifier } from "../../utils/sql-identifier"; +import { QUEUE_NOW_SQL, requireQueueTable } from "./queue-utils"; +/** + * Mark a claimed job as done: deletes it, or sets it to `done` when the + * table has `keep_completed`. + * + * Fails when the lock token no longer matches, i.e. the lease expired and + * another worker reclaimed the job. + */ +export default async function QueueAck({ table, id, lock_token, config, }) { + try { + const options = requireQueueTable(table); + const tableSql = quoteIdentifier(table); + const claimedWhere = `WHERE id = ? AND queue_status = 'processing' AND queue_lock_token = ?`; + if (!options.keep_completed) { + const res = await dbHandler({ + query: `DELETE FROM ${tableSql} ${claimedWhere} RETURNING *`, + values: [id, lock_token], + config, + }); + if (!res.success) { + return { ...res, success: false }; + } + if (!res.payload?.length) { + return { + ...res, + success: false, + msg: "Job is not claimed with this lock token", + }; + } + return res; + } + const res = await dbHandler({ + query: `UPDATE ${tableSql} SET queue_status = 'done', queue_lease_expires_at = NULL, queue_lock_token = NULL, updated_at = ${QUEUE_NOW_SQL} ${claimedWhere}`, + values: [id, lock_token], + config, + }); + if (!res.success) { + return { ...res, success: false }; + } + if (!res.db_res?.affectedRows) { + return { + ...res, + success: false, + msg: "Job is not claimed with this lock token", + }; + } + return await dbHandler({ + query: `SELECT * FROM ${tableSql} WHERE id = ?`, + values: [id], + config, + }); + } + catch (error) { + return { + success: false, + error: error.message, + }; + } +} diff --git a/dist/lib/queue/queue-cancel.d.ts b/dist/lib/queue/queue-cancel.d.ts new file mode 100644 index 0000000..41488a1 --- /dev/null +++ b/dist/lib/queue/queue-cancel.d.ts @@ -0,0 +1,12 @@ +import type { BunMariaDBConfig, DBResponseObject } from "../../types"; +type Params
= { + table: Table; + id: number | string; + config?: BunMariaDBConfig; +}; +/** + * Take a `pending` job out of the queue without deleting it. Jobs a worker + * has already claimed can't be cancelled. + */ +export default function QueueCancel
({ table, id, config, }: Params
): Promise; +export {}; diff --git a/dist/lib/queue/queue-cancel.js b/dist/lib/queue/queue-cancel.js new file mode 100644 index 0000000..ded9aaf --- /dev/null +++ b/dist/lib/queue/queue-cancel.js @@ -0,0 +1,39 @@ +import dbHandler from "../db-handler"; +import { quoteIdentifier } from "../../utils/sql-identifier"; +import { QUEUE_NOW_SQL, requireQueueTable } from "./queue-utils"; +/** + * Take a `pending` job out of the queue without deleting it. Jobs a worker + * has already claimed can't be cancelled. + */ +export default async function QueueCancel({ table, id, config, }) { + try { + requireQueueTable(table); + const tableSql = quoteIdentifier(table); + const res = await dbHandler({ + query: `UPDATE ${tableSql} SET queue_status = 'cancelled', updated_at = ${QUEUE_NOW_SQL} WHERE id = ? AND queue_status = 'pending'`, + values: [id], + config, + }); + if (!res.success) { + return { ...res, success: false }; + } + if (!res.db_res?.affectedRows) { + return { + ...res, + success: false, + msg: "Job is not pending", + }; + } + return await dbHandler({ + query: `SELECT * FROM ${tableSql} WHERE id = ?`, + values: [id], + config, + }); + } + catch (error) { + return { + success: false, + error: error.message, + }; + } +} diff --git a/dist/lib/queue/queue-claim.d.ts b/dist/lib/queue/queue-claim.d.ts new file mode 100644 index 0000000..3bf9276 --- /dev/null +++ b/dist/lib/queue/queue-claim.d.ts @@ -0,0 +1,25 @@ +import type { BunMariaDBConfig, DBResponseObject } from "../../types"; +type Params
= { + table: Table; + /** Max jobs to claim. Defaults to 1 */ + limit?: number; + /** How long the claim lasts before another worker may take the job. Defaults to the table's `lease_ms` */ + lease_ms?: number; + /** Jobs whose lease expired after this many attempts are marked `failed`. Defaults to the table's `max_attempts` */ + max_attempts?: number; + config?: BunMariaDBConfig; +}; +/** + * 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 function QueueClaim({ table, limit, lease_ms, max_attempts, config, }: Params
): Promise>; +export {}; diff --git a/dist/lib/queue/queue-claim.js b/dist/lib/queue/queue-claim.js new file mode 100644 index 0000000..8ac98ff --- /dev/null +++ b/dist/lib/queue/queue-claim.js @@ -0,0 +1,79 @@ +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, + }; + } +} diff --git a/dist/lib/queue/queue-extend.d.ts b/dist/lib/queue/queue-extend.d.ts new file mode 100644 index 0000000..bb9d73a --- /dev/null +++ b/dist/lib/queue/queue-extend.d.ts @@ -0,0 +1,16 @@ +import type { BunMariaDBConfig, DBResponseObject } from "../../types"; +type Params
= { + table: Table; + id: number | string; + /** `queue_lock_token` from the claimed row */ + lock_token: string; + /** New lease length, counted from now. Defaults to the table's `lease_ms` */ + lease_ms?: number; + config?: BunMariaDBConfig; +}; +/** + * Extend the lease on a claimed job, for jobs that run longer than the + * lease given to `claim`. + */ +export default function QueueExtend
({ table, id, lock_token, lease_ms, config, }: Params
): Promise; +export {}; diff --git a/dist/lib/queue/queue-extend.js b/dist/lib/queue/queue-extend.js new file mode 100644 index 0000000..749d1ec --- /dev/null +++ b/dist/lib/queue/queue-extend.js @@ -0,0 +1,39 @@ +import dbHandler from "../db-handler"; +import { quoteIdentifier } from "../../utils/sql-identifier"; +import { QUEUE_NOW_SQL, requireQueueTable } from "./queue-utils"; +/** + * Extend the lease on a claimed job, for jobs that run longer than the + * lease given to `claim`. + */ +export default async function QueueExtend({ table, id, lock_token, lease_ms, config, }) { + try { + const options = requireQueueTable(table); + const leaseMs = Math.max(0, Math.floor(Number(lease_ms ?? options.lease_ms) || 0)); + const res = await dbHandler({ + query: `UPDATE ${quoteIdentifier(table)} SET queue_lease_expires_at = ${QUEUE_NOW_SQL} + ?, updated_at = ${QUEUE_NOW_SQL} WHERE id = ? AND queue_status = 'processing' AND queue_lock_token = ?`, + values: [leaseMs, id, lock_token], + config, + }); + if (!res.success) { + return { ...res, success: false }; + } + if (!res.db_res?.affectedRows) { + return { + ...res, + success: false, + msg: "Job is not claimed with this lock token", + }; + } + return await dbHandler({ + query: `SELECT * FROM ${quoteIdentifier(table)} WHERE id = ?`, + values: [id], + config, + }); + } + catch (error) { + return { + success: false, + error: error.message, + }; + } +} diff --git a/dist/lib/queue/queue-fail.d.ts b/dist/lib/queue/queue-fail.d.ts new file mode 100644 index 0000000..fdcbb1f --- /dev/null +++ b/dist/lib/queue/queue-fail.d.ts @@ -0,0 +1,19 @@ +import type { BunMariaDBConfig, DBResponseObject } from "../../types"; +type Params
= { + table: Table; + id: number | string; + /** `queue_lock_token` from the claimed row */ + lock_token: string; + error?: string; + /** Wait this long before the job can be retried. Defaults to 0 */ + retry_delay_ms?: number; + /** Mark the job `failed` instead of retrying once it reaches this many attempts. Defaults to the table's `max_attempts` */ + max_attempts?: number; + config?: BunMariaDBConfig; +}; +/** + * Release a claimed job after a failed attempt. It goes back to `pending` + * (after `retry_delay_ms`) or to `failed` once `max_attempts` is reached. + */ +export default function QueueFail
({ table, id, lock_token, error, retry_delay_ms, max_attempts, config, }: Params
): Promise; +export {}; diff --git a/dist/lib/queue/queue-fail.js b/dist/lib/queue/queue-fail.js new file mode 100644 index 0000000..3d3139e --- /dev/null +++ b/dist/lib/queue/queue-fail.js @@ -0,0 +1,41 @@ +import dbHandler from "../db-handler"; +import { quoteIdentifier } from "../../utils/sql-identifier"; +import { QUEUE_NOW_SQL, requireQueueTable } from "./queue-utils"; +/** + * Release a claimed job after a failed attempt. It goes back to `pending` + * (after `retry_delay_ms`) or to `failed` once `max_attempts` is reached. + */ +export default async function QueueFail({ table, id, lock_token, error, retry_delay_ms = 0, max_attempts, config, }) { + try { + const options = requireQueueTable(table); + const tableSql = quoteIdentifier(table); + const retryDelayMs = Math.max(0, Math.floor(Number(retry_delay_ms) || 0)); + const maxAttempts = Math.max(1, Math.floor(Number(max_attempts ?? options.max_attempts) || 1)); + const res = await dbHandler({ + query: `UPDATE ${tableSql} SET queue_status = IF(queue_attempts >= ?, 'failed', 'pending'), queue_available_at = ${QUEUE_NOW_SQL} + ?, queue_lease_expires_at = NULL, queue_lock_token = NULL, queue_last_error = ?, updated_at = ${QUEUE_NOW_SQL} WHERE id = ? AND queue_status = 'processing' AND queue_lock_token = ?`, + values: [maxAttempts, retryDelayMs, error ?? null, id, lock_token], + config, + }); + if (!res.success) { + return { ...res, success: false }; + } + if (!res.db_res?.affectedRows) { + return { + ...res, + success: false, + msg: "Job is not claimed with this lock token", + }; + } + return await dbHandler({ + query: `SELECT * FROM ${quoteIdentifier(table)} WHERE id = ?`, + values: [id], + config, + }); + } + catch (error) { + return { + success: false, + error: error.message, + }; + } +} diff --git a/dist/lib/queue/queue-push.d.ts b/dist/lib/queue/queue-push.d.ts new file mode 100644 index 0000000..cfa3810 --- /dev/null +++ b/dist/lib/queue/queue-push.d.ts @@ -0,0 +1,21 @@ +import type { BunMariaDBConfig, DBResponseObject } from "../../types"; +type Params = { + table: Table; + data: Schema[]; + /** Wait this long before the jobs can be claimed */ + delay_ms?: number; + config?: BunMariaDBConfig; +}; +/** + * Add jobs to a queue table. + */ +export default function QueuePush({ table, data, delay_ms, config, }: Params): Promise; +export {}; diff --git a/dist/lib/queue/queue-push.js b/dist/lib/queue/queue-push.js new file mode 100644 index 0000000..7a8126e --- /dev/null +++ b/dist/lib/queue/queue-push.js @@ -0,0 +1,11 @@ +import DbInsert from "../mariadb/db-insert"; +/** + * Add jobs to a queue table. + */ +export default async function QueuePush({ table, data, delay_ms, config, }) { + const jobs = data.map((d) => ({ + ...(delay_ms ? { queue_available_at: Date.now() + delay_ms } : {}), + ...d, + })); + return DbInsert({ table, data: jobs, config }); +} diff --git a/dist/lib/queue/queue-utils.d.ts b/dist/lib/queue/queue-utils.d.ts new file mode 100644 index 0000000..b0b5f5f --- /dev/null +++ b/dist/lib/queue/queue-utils.d.ts @@ -0,0 +1,34 @@ +import { type BUN_MARIADB_QueueTableOptions } from "../../types"; +/** + * Current time in ms from the database clock. Leases are set and checked + * with this so workers on hosts with drifting clocks agree on expiry. + */ +export declare const QUEUE_NOW_SQL = "CAST(UNIX_TIMESTAMP(NOW(3)) * 1000 AS SIGNED)"; +/** + * Extra WHERE condition for generic update/delete on queue tables: rows a + * worker has claimed are only changed through the queue API. + */ +export declare const QUEUE_NOT_CLAIMED_SQL = "`queue_status` <> 'processing'"; +export type ResolvedQueueTableOptions = Required; +/** + * A table's `queue_table` options with defaults filled in (looked up + * directly or through its parent table), or `undefined` if it is not a + * queue table. + */ +export declare function grabQueueTableOptions(tableName: string): ResolvedQueueTableOptions | undefined; +/** + * Whether a table is declared with `queue_table` (directly or through its + * parent table). + */ +export declare function isQueueTable(tableName: string): boolean; +/** + * Queue options for a table the queue API was called on. Throws if the table + * is not a queue table. + */ +export declare function requireQueueTable(tableName: string): ResolvedQueueTableOptions; +/** + * Throw if `data` sets queue state the queue API owns. + */ +export declare function assertNoProtectedQueueFields(table: string, data: { + [k: string]: any; +}[]): void; diff --git a/dist/lib/queue/queue-utils.js b/dist/lib/queue/queue-utils.js new file mode 100644 index 0000000..926f04d --- /dev/null +++ b/dist/lib/queue/queue-utils.js @@ -0,0 +1,69 @@ +import _ from "lodash"; +import { readLiveSchema } from "../../functions/live-schema"; +import { QueueProtectedFields, } from "../../types"; +/** + * Current time in ms from the database clock. Leases are set and checked + * with this so workers on hosts with drifting clocks agree on expiry. + */ +export const QUEUE_NOW_SQL = "CAST(UNIX_TIMESTAMP(NOW(3)) * 1000 AS SIGNED)"; +/** + * Extra WHERE condition for generic update/delete on queue tables: rows a + * worker has claimed are only changed through the queue API. + */ +export const QUEUE_NOT_CLAIMED_SQL = "`queue_status` <> 'processing'"; +const QUEUE_TABLE_DEFAULTS = { + max_attempts: 5, + lease_ms: 30_000, + keep_completed: false, +}; +/** + * A table's `queue_table` options with defaults filled in (looked up + * directly or through its parent table), or `undefined` if it is not a + * queue table. + */ +export function grabQueueTableOptions(tableName) { + const dbSchema = global.DB_SCHEMA || readLiveSchema(); + const table = dbSchema?.tables?.find((t) => t.tableName === tableName); + if (!table) + return undefined; + if (table.queue_table === undefined) { + return table.parentTableName + ? grabQueueTableOptions(table.parentTableName) + : undefined; + } + if (!table.queue_table) + return undefined; + return { + ...QUEUE_TABLE_DEFAULTS, + ...(typeof table.queue_table === "object" + ? _.omitBy(table.queue_table, _.isUndefined) + : {}), + }; +} +/** + * Whether a table is declared with `queue_table` (directly or through its + * parent table). + */ +export function isQueueTable(tableName) { + return Boolean(grabQueueTableOptions(tableName)); +} +/** + * Queue options for a table the queue API was called on. Throws if the table + * is not a queue table. + */ +export function requireQueueTable(tableName) { + const options = grabQueueTableOptions(tableName); + if (!options) { + throw new Error(`\`${tableName}\` is not a queue table`); + } + return options; +} +/** + * Throw if `data` sets queue state the queue API owns. + */ +export function assertNoProtectedQueueFields(table, data) { + const protectedFields = QueueProtectedFields.filter((field) => data.some((d) => d && d[field] !== undefined)); + if (protectedFields.length > 0) { + throw new Error(`\`${table}\` is a queue table: ${protectedFields.join(", ")} can only be changed through BunMariaDB.queue`); + } +} diff --git a/dist/lib/schema/build-column-definition.js b/dist/lib/schema/build-column-definition.js index cbd0ab8..47fd7ab 100644 --- a/dist/lib/schema/build-column-definition.js +++ b/dist/lib/schema/build-column-definition.js @@ -1,6 +1,7 @@ import isVectorField from "./is-vector-field"; import mapDataType from "./map-data-types"; import MariaDBQuoteGen from "./mariadb-quote-gen"; +import { escapeStringLiteral } from "../../utils/sql-identifier"; export default function buildColumnDefinition(field, options = {}) { if (!field.fieldName) { throw new Error("Field name is required"); @@ -27,7 +28,7 @@ export default function buildColumnDefinition(field, options = {}) { } if (field.defaultValue !== undefined) { if (typeof field.defaultValue === "string") { - parts.push(`DEFAULT '${field.defaultValue.replace(/'/g, "''")}'`); + parts.push(`DEFAULT '${escapeStringLiteral(field.defaultValue)}'`); } else { parts.push(`DEFAULT ${field.defaultValue}`); diff --git a/dist/lib/schema/map-data-types.js b/dist/lib/schema/map-data-types.js index a50ba84..542773d 100644 --- a/dist/lib/schema/map-data-types.js +++ b/dist/lib/schema/map-data-types.js @@ -1,3 +1,4 @@ +import { escapeStringLiteral } from "../../utils/sql-identifier"; export default function mapDataType(field) { const dataType = field.dataType?.toUpperCase() || "TEXT"; const vectorSize = field.vectorSize || 1536; @@ -76,13 +77,13 @@ export default function mapDataType(field) { return "TINYINT(1)"; case "ENUM": { const enumVals = (field.options || []) - .map((v) => `'${String(v).replace(/'/g, "''")}'`) + .map((v) => `'${escapeStringLiteral(String(v))}'`) .join(", "); return `ENUM(${enumVals || "''"})`; } case "SET": { const setVals = (field.options || []) - .map((v) => `'${String(v).replace(/'/g, "''")}'`) + .map((v) => `'${escapeStringLiteral(String(v))}'`) .join(", "); return `SET(${setVals || "''"})`; } diff --git a/dist/lib/schema/update-table.js b/dist/lib/schema/update-table.js index 24eb8ff..9b5ab63 100644 --- a/dist/lib/schema/update-table.js +++ b/dist/lib/schema/update-table.js @@ -12,8 +12,8 @@ import { dropForeignKeysOnColumns } from "./sync-foreign-keys"; * Live types often include display widths (e.g. bigint(20) vs BIGINT). */ function columnTypesMatch(liveType, expectedType) { - const live = liveType.toLowerCase().replace(/\s+/g, ""); - const expected = expectedType.toLowerCase().replace(/\s+/g, ""); + const live = normalizeSqlOutsideQuotes(liveType, ""); + const expected = normalizeSqlOutsideQuotes(expectedType, ""); if (live === expected) return true; // live may include display width: bigint(20) vs bigint @@ -41,25 +41,71 @@ function vectorTypeDiverged(liveType, liveComment, field) { } return true; } -function normalizeDefault(value) { +/** + * Lowercase SQL and collapse whitespace to `space`, leaving the contents of + * single-quoted literals untouched (ENUM options, string defaults). + */ +function normalizeSqlOutsideQuotes(sql, space) { + return sql + .trim() + .split(/('(?:[^'\\]|''|\\.)*')/) + .map((part, i) => i % 2 === 1 ? part : part.toLowerCase().replace(/\s+/g, space)) + .join(""); +} +/** + * Parse a default as MariaDB reports it in COLUMN_DEFAULT (or as written in + * `defaultValueLiteral`): unquoted NULL, an escaped quoted string literal, + * or an expression / number. + */ +function parseDefault(value) { if (value == null) - return ""; - let v = String(value).trim(); - // Strip surrounding quotes MariaDB may include - if ((v.startsWith("'") && v.endsWith("'")) || - (v.startsWith('"') && v.endsWith('"'))) { - v = v.slice(1, -1); + return { kind: "none" }; + const v = String(value).trim(); + if (/^null$/i.test(v)) + return { kind: "none" }; + if (/^'(?:[^'\\]|''|\\.)*'$/s.test(v)) { + const escapes = { + n: "\n", + t: "\t", + r: "\r", + "0": "\0", + }; + return { + kind: "string", + value: v + .slice(1, -1) + .replace(/''|\\(.)/gs, (m, ch) => m === "''" ? "'" : (escapes[ch] ?? ch)), + }; } - return v.toLowerCase().replace(/\s+/g, " "); + let expr = normalizeSqlOutsideQuotes(v, ""); + // MariaDB may wrap expressions in parentheses + while (/^\(.*\)$/.test(expr) && balancedParens(expr.slice(1, -1))) { + expr = expr.slice(1, -1); + } + return { kind: "expr", value: expr }; +} +function balancedParens(s) { + let depth = 0; + for (const ch of s) { + if (ch === "(") + depth++; + if (ch === ")" && --depth < 0) + return false; + } + return depth === 0; } function expectedDefault(field) { if (field.defaultValue !== undefined) { - return String(field.defaultValue); + // Strings are emitted as quoted literals; numbers verbatim + if (typeof field.defaultValue === "string") { + return { kind: "string", value: field.defaultValue }; + } + return parseDefault(String(field.defaultValue)); } if (field.defaultValueLiteral) { - return field.defaultValueLiteral; + return parseDefault(field.defaultValueLiteral); } - return null; + return { kind: "none" }; } function expectedOnUpdate(field) { if (field.onUpdate) @@ -68,15 +114,29 @@ function expectedOnUpdate(field) { return field.onUpdateLiteral; return null; } -function defaultsMatch(liveDefault, expected) { - if (liveDefault === expected) - return true; - if (expected.includes("current_timestamp") && - liveDefault.includes("current_timestamp")) { - return true; +function defaultsMatch(live, expected) { + if (live.kind === "none" || expected.kind === "none") { + return live.kind === expected.kind; + } + if (live.value === expected.value) { + // A string never equals an expression of the same text, except for + // numbers, which MariaDB reports unquoted on numeric columns + return live.kind === expected.kind || isNumeric(live.value); + } + // DECIMAL defaults are reported padded: 1.5 → 1.50 + if (isNumeric(live.value) && isNumeric(expected.value)) { + return Number(live.value) === Number(expected.value); + } + if (live.kind === "expr" && expected.kind === "expr") { + // now() / current_timestamp / current_timestamp(N) are aliases + const ts = /^(current_timestamp|now|localtimestamp|localtime)(\(\d*\))?$/; + return ts.test(live.value) && ts.test(expected.value); } return false; } +function isNumeric(value) { + return value.trim() !== "" && Number.isFinite(Number(value)); +} function columnAttributesDiverged(live, field) { const wantsNotNull = fieldRequiresNotNull(field); if (wantsNotNull && live.isNullable) { @@ -92,12 +152,12 @@ function columnAttributesDiverged(live, field) { return true; } const wantsDefault = expectedDefault(field); - if (wantsDefault != null) { - const liveDefault = normalizeDefault(live.columnDefault); - const expected = normalizeDefault(wantsDefault); - if (!defaultsMatch(liveDefault, expected)) { - return true; - } + const liveDefault = parseDefault(live.columnDefault); + // Older servers give TIMESTAMP columns an implicit default + const implicitDefault = wantsDefault.kind === "none" && + field.dataType?.toUpperCase() === "TIMESTAMP"; + if (!implicitDefault && !defaultsMatch(liveDefault, wantsDefault)) { + return true; } const wantsOnUpdate = expectedOnUpdate(field); const hasOnUpdate = /on update/i.test(live.extra || ""); @@ -108,11 +168,9 @@ function columnAttributesDiverged(live, field) { return true; } if (wantsOnUpdate && hasOnUpdate) { - const liveOnUpdate = normalizeDefault((live.extra.match(/on update\s+(.+)/i) || [])[1] || ""); - const expectedOu = normalizeDefault(wantsOnUpdate); - if (liveOnUpdate && - expectedOu && - !defaultsMatch(liveOnUpdate, expectedOu)) { + const liveOnUpdate = parseDefault((live.extra.match(/on update\s+(.+)/i) || [])[1] || ""); + const expectedOu = parseDefault(wantsOnUpdate); + if (!defaultsMatch(liveOnUpdate, expectedOu)) { return true; } } diff --git a/dist/types/index.d.ts b/dist/types/index.d.ts index d4d615f..97c9b4f 100644 --- a/dist/types/index.d.ts +++ b/dist/types/index.d.ts @@ -74,6 +74,24 @@ export interface BUN_MARIADB_TableSchemaType { */ childTableDbId?: string | number; collation?: (typeof MariaDBCollations)[number]; + /** + * Make this table a job queue. Adds the `QueueFields` columns and + * `QueueIndexes`, and routes state changes through `BunMariaDB.queue`. + * Pass an options object to set the queue's defaults. + */ + queue_table?: boolean | BUN_MARIADB_QueueTableOptions; +} +/** + * Defaults for a queue table. Per-call arguments to `BunMariaDB.queue` + * override them. + */ +export interface BUN_MARIADB_QueueTableOptions { + /** Attempts before a job is marked `failed`. Defaults to 5 */ + max_attempts?: number; + /** How long a claim lasts before another worker may take the job. Defaults to 30s */ + lease_ms?: number; + /** `ack` marks jobs `done` and keeps them instead of deleting them */ + keep_completed?: boolean; } /** * Reference object used to link a table to one of its child tables. @@ -1425,6 +1443,28 @@ export type BunMariaDBConfigReturn = { * Default fields automatically suggested for new tables. */ export declare const DefaultFields: BUN_MARIADB_FieldSchemaType[]; +/** + * Lifecycle of a queue row. Only `pending` jobs are claimed. + * - `done`: acknowledged, kept because of `keep_completed` (otherwise deleted) + * - `failed`: used up its attempts (dead letter) + * - `cancelled`: removed from the queue with `queue.cancel` + */ +export declare const QueueStatuses: readonly ["pending", "processing", "done", "failed", "cancelled"]; +export type QueueStatus = (typeof QueueStatuses)[number]; +/** + * Fields added to every table with `queue_table: true`. + */ +export declare const QueueFields: BUN_MARIADB_FieldSchemaType[]; +/** + * Queue fields only the queue API may write. `queue_available_at` can be set + * on insert and update. + */ +export declare const QueueProtectedFields: readonly ["queue_status", "queue_attempts", "queue_lease_expires_at", "queue_lock_token", "queue_last_error"]; +/** + * Indexes added to queue tables. Claims scan `idx_queue_claim` in order, so + * `LIMIT` stops the scan early and only the claimed rows get locked. + */ +export declare const QueueIndexes: BUN_MARIADB_IndexSchemaType[]; export type BunMariaDBQueryFieldValues = { field: F; table?: T; diff --git a/dist/types/index.js b/dist/types/index.js index 4a89e5a..fe3af24 100644 --- a/dist/types/index.js +++ b/dist/types/index.js @@ -235,6 +235,87 @@ export const DefaultFields = [ fieldDescription: "The time when the record was updated. (Unix Timestamp)", }, ]; +/** + * Lifecycle of a queue row. Only `pending` jobs are claimed. + * - `done`: acknowledged, kept because of `keep_completed` (otherwise deleted) + * - `failed`: used up its attempts (dead letter) + * - `cancelled`: removed from the queue with `queue.cancel` + */ +export const QueueStatuses = [ + "pending", + "processing", + "done", + "failed", + "cancelled", +]; +/** + * Fields added to every table with `queue_table: true`. + */ +export const QueueFields = [ + { + fieldName: "queue_status", + dataType: "ENUM", + options: [...QueueStatuses], + notNullValue: true, + defaultValue: "pending", + fieldDescription: "Queue state of the job.", + }, + { + fieldName: "queue_available_at", + dataType: "BIGINT", + notNullValue: true, + defaultValue: 0, + fieldDescription: "The job cannot be claimed before this time. (Unix Timestamp, ms)", + }, + { + fieldName: "queue_attempts", + dataType: "INT", + notNullValue: true, + defaultValue: 0, + fieldDescription: "How many times the job has been claimed.", + }, + { + fieldName: "queue_lease_expires_at", + dataType: "BIGINT", + fieldDescription: "When the current claim expires and the job can be reclaimed. (Unix Timestamp, ms)", + }, + { + fieldName: "queue_lock_token", + dataType: "VARCHAR", + dataLength: 64, + fieldDescription: "Token of the current claim. Required to ack, fail or extend the job.", + }, + { + fieldName: "queue_last_error", + dataType: "TEXT", + fieldDescription: "Error message from the last failed attempt.", + }, +]; +/** + * Queue fields only the queue API may write. `queue_available_at` can be set + * on insert and update. + */ +export const QueueProtectedFields = [ + "queue_status", + "queue_attempts", + "queue_lease_expires_at", + "queue_lock_token", + "queue_last_error", +]; +/** + * Indexes added to queue tables. Claims scan `idx_queue_claim` in order, so + * `LIMIT` stops the scan early and only the claimed rows get locked. + */ +export const QueueIndexes = [ + { + indexName: "idx_queue_claim", + indexTableFields: ["queue_status", "queue_available_at"], + }, + { + indexName: "idx_queue_lease", + indexTableFields: ["queue_status", "queue_lease_expires_at"], + }, +]; export const RequiredENVs = [ "BUN_MARIADB_SERVER_HOST", "BUN_MARIADB_SERVER_PORT", diff --git a/dist/utils/append-default-fields-to-db-schema.js b/dist/utils/append-default-fields-to-db-schema.js index d1b758a..1ee8bdc 100644 --- a/dist/utils/append-default-fields-to-db-schema.js +++ b/dist/utils/append-default-fields-to-db-schema.js @@ -1,11 +1,31 @@ import _ from "lodash"; -import { DefaultFields } from "../types"; +import { DefaultFields, QueueFields, QueueIndexes, } from "../types"; +const queueFieldNames = new Set(QueueFields.map((f) => f.fieldName)); +const queueIndexNames = new Set(QueueIndexes.map((i) => i.indexName)); export default function ({ dbSchema }) { const finaldbSchema = _.cloneDeep(dbSchema); finaldbSchema.tables = finaldbSchema.tables.map((t) => { const newTable = _.cloneDeep(t); newTable.fields = newTable.fields.filter((f) => !f.fieldName?.match(/^(id|created_at|updated_at)$/)); newTable.fields.unshift(...DefaultFields); + if (newTable.queue_table) { + const reserved = [ + ...newTable.fields + .map((f) => f.fieldName) + .filter((name) => queueFieldNames.has(name)), + ...(newTable.indexes || []) + .map((i) => i.indexName) + .filter((name) => queueIndexNames.has(name)), + ]; + if (reserved.length > 0) { + throw new Error(`\`${newTable.tableName}\` is a queue table: ${reserved.join(", ")} ${reserved.length > 1 ? "are" : "is"} managed by bun-mariadb and can't be declared in the schema`); + } + newTable.fields.push(..._.cloneDeep(QueueFields)); + newTable.indexes = [ + ...(newTable.indexes || []), + ..._.cloneDeep(QueueIndexes), + ]; + } return newTable; }); return finaldbSchema;