tbenandClaude Opus 5.5
b8dd453519
grabMariaDBClient reused global.MARIADB_CLIENT whenever the requested config matched global.CONFIG. It never checked that the client was made from that config or was still open. init() sets global.CONFIG before creating the client, so a re-init after a close, or a config change, got back a closed or wrong client. - Remember the cache key each client was created with, and compare that - Wrapping close() drops the client from the cache and unsets global.MARIADB_CLIENT, so the next grab or ensureInit makes a new one - resolveClient goes through grabMariaDBClient instead of repeating the global.CONFIG comparison - Remove the queue test's workaround for this Co-Authored-By: Claude Opus 5.5 <[email protected]>
249 lines
11 KiB
TypeScript
249 lines
11 KiB
TypeScript
/**
|
|
* Queue table integration tests against a real MariaDB.
|
|
*
|
|
* Skipped unless BUN_MARIADB_TEST_DB is set (see tests/db.test.ts).
|
|
*/
|
|
import { afterAll, beforeAll, beforeEach, describe, expect, test } from "bun:test";
|
|
import type { BUN_MARIADB_DatabaseSchemaType } from "../src/types";
|
|
|
|
const TEST_DB = process.env.BUN_MARIADB_TEST_DB;
|
|
const TABLE = "bmt_jobs";
|
|
const KEPT_TABLE = "bmt_jobs_kept";
|
|
|
|
const schema: BUN_MARIADB_DatabaseSchemaType = {
|
|
tables: [
|
|
{
|
|
tableName: TABLE,
|
|
queue_table: true,
|
|
fields: [{ fieldName: "name", dataType: "VARCHAR" }],
|
|
},
|
|
{
|
|
tableName: KEPT_TABLE,
|
|
queue_table: { keep_completed: true, max_attempts: 2 },
|
|
fields: [{ fieldName: "name", dataType: "VARCHAR" }],
|
|
},
|
|
],
|
|
} as BUN_MARIADB_DatabaseSchemaType;
|
|
|
|
test("queue tables refuse user-declared queue fields and indexes", async () => {
|
|
const appendDefaults = (await import("../src/utils/append-default-fields-to-db-schema")).default;
|
|
|
|
expect(() =>
|
|
appendDefaults({
|
|
dbSchema: {
|
|
tables: [
|
|
{
|
|
tableName: "q",
|
|
queue_table: true,
|
|
fields: [{ fieldName: "queue_status", dataType: "VARCHAR" }],
|
|
indexes: [{ indexName: "idx_queue_claim", indexTableFields: ["id"] }],
|
|
},
|
|
],
|
|
},
|
|
}),
|
|
).toThrow("queue_status, idx_queue_claim are managed by bun-mariadb");
|
|
});
|
|
|
|
describe.skipIf(!TEST_DB)("Queue tables", () => {
|
|
let DB: typeof import("../src/index").default;
|
|
|
|
beforeAll(async () => {
|
|
const setMariaDBClient = (await import("../src/functions/set-mariadb-client")).default;
|
|
const createDBSchema = (await import("../src/lib/schema/create-db-schema")).default;
|
|
const appendDefaults = (await import("../src/utils/append-default-fields-to-db-schema")).default;
|
|
|
|
// The test container uses a self-signed certificate
|
|
global.CONFIG = { db_name: TEST_DB!, ssl_reject_unauthorized: false };
|
|
global.DB_SCHEMA = schema;
|
|
setMariaDBClient({ config: global.CONFIG });
|
|
|
|
DB = (await import("../src/index")).default;
|
|
|
|
await createDBSchema({
|
|
db_schema: appendDefaults({ dbSchema: schema }),
|
|
config: global.CONFIG,
|
|
});
|
|
});
|
|
|
|
beforeEach(async () => {
|
|
await DB.sql({ sql: `DELETE FROM ${TABLE}` });
|
|
await DB.sql({ sql: `DELETE FROM ${KEPT_TABLE}` });
|
|
});
|
|
|
|
afterAll(async () => {
|
|
await DB?.sql({ sql: `DROP TABLE IF EXISTS ${TABLE}, ${KEPT_TABLE}` });
|
|
await DB?.sql({
|
|
sql: `DELETE FROM __db_schema_manager__ WHERE table_name IN ('${TABLE}', '${KEPT_TABLE}')`,
|
|
});
|
|
await global.MARIADB_CLIENT?.close();
|
|
});
|
|
|
|
test("schema adds queue columns and indexes", async () => {
|
|
const cols = await DB.sql<any>({
|
|
sql: "SELECT COLUMN_NAME FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ?",
|
|
values: [TABLE],
|
|
});
|
|
const names = cols.payload?.map((c) => c.COLUMN_NAME);
|
|
expect(names).toContain("queue_status");
|
|
expect(names).toContain("queue_lock_token");
|
|
|
|
const idx = await DB.sql<any>({
|
|
sql: "SELECT COLUMN_NAME FROM information_schema.STATISTICS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND INDEX_NAME = 'idx_queue_claim' ORDER BY SEQ_IN_INDEX",
|
|
values: [TABLE],
|
|
});
|
|
expect(idx.payload?.map((c) => c.COLUMN_NAME)).toEqual(["queue_status", "queue_available_at"]);
|
|
});
|
|
|
|
test("insert and update refuse protected queue fields", async () => {
|
|
const ins = await DB.insert({ table: TABLE, data: [{ name: "x", queue_status: "processing" }] });
|
|
expect(ins.success).toBe(false);
|
|
expect(ins.error).toContain("queue_status");
|
|
|
|
await DB.queue.push({ table: TABLE, data: [{ name: "a" }] });
|
|
const upd = await DB.update({ table: TABLE, query: { query: { name: { value: "a" } } }, data: { queue_attempts: 0 } });
|
|
expect(upd.success).toBe(false);
|
|
});
|
|
|
|
test("claim, then ack with the lock token deletes the job", async () => {
|
|
await DB.queue.push({ table: TABLE, data: [{ name: "a" }] });
|
|
|
|
const claim = await DB.queue.claim<any>({ table: TABLE });
|
|
expect(claim.success).toBe(true);
|
|
const job = claim.payload![0];
|
|
expect(job.queue_status).toBe("processing");
|
|
expect(job.queue_attempts).toBe(1);
|
|
expect(job.queue_lock_token).toBeTruthy();
|
|
|
|
expect((await DB.queue.claim({ table: TABLE })).payload).toEqual([]);
|
|
|
|
const bad = await DB.queue.ack({ table: TABLE, id: job.id, lock_token: "nope" });
|
|
expect(bad.success).toBe(false);
|
|
|
|
const ack = await DB.queue.ack({ table: TABLE, id: job.id, lock_token: job.queue_lock_token });
|
|
expect(ack.success).toBe(true);
|
|
expect((await DB.select({ table: TABLE })).payload).toEqual([]);
|
|
});
|
|
|
|
test("concurrent claims never return the same job twice", async () => {
|
|
await DB.queue.push({ table: TABLE, data: Array.from({ length: 20 }, (_, i) => ({ name: `j${i}` })) });
|
|
|
|
const claims = await Promise.all(
|
|
Array.from({ length: 8 }, () => DB.queue.claim<any>({ table: TABLE, limit: 3 })),
|
|
);
|
|
|
|
expect(claims.every((c) => c.success)).toBe(true);
|
|
const ids = claims.flatMap((c) => c.payload!.map((j) => j.id));
|
|
expect(new Set(ids).size).toBe(ids.length);
|
|
expect(ids.length).toBe(20);
|
|
});
|
|
|
|
test("update and delete skip claimed jobs", async () => {
|
|
await DB.queue.push({ table: TABLE, data: [{ name: "a" }] });
|
|
const job = (await DB.queue.claim<any>({ table: TABLE })).payload![0];
|
|
|
|
const upd = await DB.update({ table: TABLE, targetId: job.id, data: { name: "b" } });
|
|
expect(upd.success).toBe(false);
|
|
|
|
await DB.delete({ table: TABLE, targetId: job.id });
|
|
expect((await DB.select({ table: TABLE })).payload?.length).toBe(1);
|
|
});
|
|
|
|
test("fail retries, then dead-letters after max_attempts", async () => {
|
|
await DB.queue.push({ table: TABLE, data: [{ name: "a" }] });
|
|
|
|
let job = (await DB.queue.claim<any>({ table: TABLE })).payload![0];
|
|
const retry = await DB.queue.fail({ table: TABLE, id: job.id, lock_token: job.queue_lock_token, error: "boom", max_attempts: 2 });
|
|
expect(retry.success).toBe(true);
|
|
expect(retry.single_res?.queue_status).toBe("pending");
|
|
expect(retry.single_res?.queue_last_error).toBe("boom");
|
|
|
|
job = (await DB.queue.claim<any>({ table: TABLE })).payload![0];
|
|
expect(job.queue_attempts).toBe(2);
|
|
const dead = await DB.queue.fail({ table: TABLE, id: job.id, lock_token: job.queue_lock_token, error: "boom", max_attempts: 2 });
|
|
expect(dead.single_res?.queue_status).toBe("failed");
|
|
|
|
expect((await DB.queue.claim({ table: TABLE })).payload).toEqual([]);
|
|
});
|
|
|
|
test("an expired lease is reclaimed and the old token stops working", async () => {
|
|
await DB.queue.push({ table: TABLE, data: [{ name: "a" }] });
|
|
|
|
const first = (await DB.queue.claim<any>({ table: TABLE, lease_ms: 0 })).payload![0];
|
|
const second = (await DB.queue.claim<any>({ table: TABLE })).payload![0];
|
|
expect(second.id).toBe(first.id);
|
|
expect(second.queue_attempts).toBe(2);
|
|
|
|
expect((await DB.queue.ack({ table: TABLE, id: first.id, lock_token: first.queue_lock_token })).success).toBe(false);
|
|
expect((await DB.queue.extend({ table: TABLE, id: second.id, lock_token: second.queue_lock_token })).success).toBe(true);
|
|
});
|
|
|
|
test("an expired lease on the last attempt is dead-lettered", async () => {
|
|
await DB.queue.push({ table: TABLE, data: [{ name: "a" }] });
|
|
await DB.queue.claim({ table: TABLE, lease_ms: 0 });
|
|
|
|
expect((await DB.queue.claim({ table: TABLE, max_attempts: 1 })).payload).toEqual([]);
|
|
const row = (await DB.select<any>({ table: TABLE })).single_res;
|
|
expect(row?.queue_status).toBe("failed");
|
|
});
|
|
|
|
test("claims in push order and holds back delayed jobs", async () => {
|
|
await DB.queue.push({ table: TABLE, data: [{ name: "later" }], delay_ms: 60_000 });
|
|
await DB.queue.push({ table: TABLE, data: [{ name: "first" }, { name: "second" }] });
|
|
|
|
const claim = await DB.queue.claim<any>({ table: TABLE, limit: 5 });
|
|
expect(claim.payload?.map((j) => j.name)).toEqual(["first", "second"]);
|
|
});
|
|
|
|
test("keep_completed: ack marks the job done and keeps it", async () => {
|
|
await DB.queue.push({ table: KEPT_TABLE, data: [{ name: "a" }] });
|
|
const job = (await DB.queue.claim<any>({ table: KEPT_TABLE })).payload![0];
|
|
|
|
const ack = await DB.queue.ack({ table: KEPT_TABLE, id: job.id, lock_token: job.queue_lock_token });
|
|
expect(ack.success).toBe(true);
|
|
expect(ack.single_res?.queue_status).toBe("done");
|
|
expect(ack.single_res?.queue_lock_token).toBeNull();
|
|
|
|
expect((await DB.queue.claim({ table: KEPT_TABLE })).payload).toEqual([]);
|
|
expect((await DB.queue.ack({ table: KEPT_TABLE, id: job.id, lock_token: job.queue_lock_token })).success).toBe(false);
|
|
});
|
|
|
|
test("table max_attempts is the default for fail", async () => {
|
|
await DB.queue.push({ table: KEPT_TABLE, data: [{ name: "a" }] });
|
|
|
|
for (const expected of ["pending", "failed"]) {
|
|
const job = (await DB.queue.claim<any>({ table: KEPT_TABLE })).payload![0];
|
|
const res = await DB.queue.fail({ table: KEPT_TABLE, id: job.id, lock_token: job.queue_lock_token });
|
|
expect(res.single_res?.queue_status).toBe(expected);
|
|
}
|
|
});
|
|
|
|
test("cancel takes pending jobs out of the queue, but not claimed ones", async () => {
|
|
await DB.queue.push({ table: TABLE, data: [{ name: "a" }, { name: "b" }] });
|
|
const rows = (await DB.select<any>({ table: TABLE, query: { order: { field: "id", strategy: "ASC" } } })).payload!;
|
|
|
|
const cancel = await DB.queue.cancel({ table: TABLE, id: rows[0].id });
|
|
expect(cancel.single_res?.queue_status).toBe("cancelled");
|
|
|
|
const claimed = (await DB.queue.claim<any>({ table: TABLE, limit: 5 })).payload!;
|
|
expect(claimed.map((j) => j.name)).toEqual(["b"]);
|
|
|
|
expect((await DB.queue.cancel({ table: TABLE, id: claimed[0].id })).success).toBe(false);
|
|
});
|
|
|
|
test("select filters queue rows by status", async () => {
|
|
await DB.queue.push({ table: TABLE, data: [{ name: "a" }, { name: "b" }, { name: "c" }] });
|
|
const job = (await DB.queue.claim<any>({ table: TABLE })).payload![0];
|
|
await DB.queue.fail({ table: TABLE, id: job.id, lock_token: job.queue_lock_token, max_attempts: 1 });
|
|
|
|
const failed = await DB.select<any>({ table: TABLE, query: { query: { queue_status: { value: "failed" } } } });
|
|
expect(failed.payload?.map((r) => r.name)).toEqual(["a"]);
|
|
|
|
const open = await DB.select<any>({
|
|
table: TABLE,
|
|
count: true,
|
|
query: { query: { queue_status: { value: ["pending", "processing"], equality: "IN" } } },
|
|
});
|
|
expect(open.count).toBe(2);
|
|
});
|
|
});
|