tbenandClaude Opus 5.5
938a1da71d
- isQueue table option adds queue columns and a (status, available_at) index; BunSQLite.queue exposes add, claim, complete, fail and release_stale with atomic claims - Set busy_timeout on connections so workers in separate processes wait instead of failing with SQLITE_BUSY - Quote table and column names with backticks in generated SQL, so reserved words (to, order, group) work and unknown columns error instead of matching as string literals - Tests for queue ops, multi-process claims and reserved-word names Co-Authored-By: Claude Opus 5.5 <[email protected]>
43 lines
1.3 KiB
TypeScript
43 lines
1.3 KiB
TypeScript
/**
|
|
* Queue worker for the multi-process claim test. Claims jobs one at a time
|
|
* until the queue is empty and prints the claimed ids as JSON.
|
|
*/
|
|
import fs from "fs";
|
|
import path from "path";
|
|
import grabSQLiteClient from "../../src/functions/grab-sqlite-client";
|
|
import { QueueClaim, QueueComplete } from "../../src/lib/sqlite/db-queue";
|
|
import type { BunSQLiteConfig, BUN_SQLITE_DatabaseSchemaType } from "../../src/types";
|
|
|
|
const config: BunSQLiteConfig = JSON.parse(process.env.QUEUE_TEST_CONFIG!);
|
|
const dbSchema: BUN_SQLITE_DatabaseSchemaType = JSON.parse(
|
|
process.env.QUEUE_TEST_SCHEMA!,
|
|
);
|
|
const worker = process.env.QUEUE_TEST_WORKER!;
|
|
|
|
global.CONFIG = config;
|
|
global.DB_SCHEMA = dbSchema;
|
|
global.SQLITE_CLIENT = grabSQLiteClient({ config });
|
|
|
|
// Signal ready, then wait for the test to start every worker at once
|
|
fs.writeFileSync(path.join(config.db_dir!, `ready-${worker}`), "");
|
|
while (!fs.existsSync(path.join(config.db_dir!, "go"))) {}
|
|
|
|
const ids: number[] = [];
|
|
|
|
while (true) {
|
|
const res = await QueueClaim({ table: "jobs", worker });
|
|
|
|
if (!res.success) {
|
|
console.error(res.error);
|
|
process.exit(1);
|
|
}
|
|
|
|
const job = res.singleRes;
|
|
if (!job) break;
|
|
|
|
ids.push(job.id);
|
|
await QueueComplete({ table: "jobs", id: job.id });
|
|
}
|
|
|
|
console.log(JSON.stringify(ids));
|