From 7c8a2c4ea89623b1e1b889d2f8aacaa665edb190 Mon Sep 17 00:00:00 2001 From: 3m1n3nc3 Date: Wed, 21 Oct 2026 00:03:24 +0100 Subject: [PATCH 1/2] style: apply prettier formatting --- create-arkstack/src/catalog.ts | 40 +++++++++---------- .../console/src/prepare/BuildInterfaces.ts | 6 +-- .../resources/js/Pages/Index.tsx | 3 +- .../app/http/controllers/BaseController.ts | 5 +-- .../express-inertia/src/config/filesystem.ts | 5 +-- templates/express-inertia/tsdown.config.ts | 5 +-- 6 files changed, 26 insertions(+), 38 deletions(-) diff --git a/create-arkstack/src/catalog.ts b/create-arkstack/src/catalog.ts index bb535e95..c6c701f8 100644 --- a/create-arkstack/src/catalog.ts +++ b/create-arkstack/src/catalog.ts @@ -4,26 +4,26 @@ /** The default workspace catalog (`catalog:`). */ export const catalog: Record = { - "@h3ravel/musket": "^2.2.13", - "@h3ravel/shared": "^2.2.7", - "@h3ravel/support": "^2.2.7", - "@h3ravel/collect.js": "^5.4.0", - "@faker-js/faker": "^10.4.0", - "@arkormx/plugin-clear-router": "^0.1.57", - "@resora/plugin-clear-router": "^1.0.72", - "@kanun-hq/plugin-file": "^0.1.12", - "@kanun-hq/plugin-phone": "^0.1.8", - "arkormx": "^2.12.8", - "clear-router": "^2.9.3", - "dotenv": "^17.4.2", - "express": "^5.2.1", - "h3": "^2.0.1-rc.22", - "kysely": "^0.29.2", - "kanun": "^1.2.0", - "parasito": "^0.2.6", - "pg": "^8.22.0", - "resora": "^1.3.34", - "tsdown": "^0.22.14" + '@h3ravel/musket': '^2.2.13', + '@h3ravel/shared': '^2.2.7', + '@h3ravel/support': '^2.2.7', + '@h3ravel/collect.js': '^5.4.0', + '@faker-js/faker': '^10.4.0', + '@arkormx/plugin-clear-router': '^0.1.57', + '@resora/plugin-clear-router': '^1.0.72', + '@kanun-hq/plugin-file': '^0.1.12', + '@kanun-hq/plugin-phone': '^0.1.8', + arkormx: '^2.12.8', + 'clear-router': '^2.9.3', + dotenv: '^17.4.2', + express: '^5.2.1', + h3: '^2.0.1-rc.22', + kysely: '^0.29.2', + kanun: '^1.2.0', + parasito: '^0.2.6', + pg: '^8.22.0', + resora: '^1.3.34', + tsdown: '^0.22.14', } /** Named workspace catalogs (`catalog:`). */ diff --git a/packages/console/src/prepare/BuildInterfaces.ts b/packages/console/src/prepare/BuildInterfaces.ts index 126b1291..8d2536f1 100644 --- a/packages/console/src/prepare/BuildInterfaces.ts +++ b/packages/console/src/prepare/BuildInterfaces.ts @@ -25,9 +25,7 @@ export class BuildInterfaces { static async tsconfig() { const pkg = JSON.parse(await readFile(path.join(process.cwd(), 'package.json'), 'utf-8')) - const deps = Object - .keys(pkg.dependencies ?? {}) - .concat(Object.keys(pkg.devDependencies ?? {})) + const deps = Object.keys(pkg.dependencies ?? {}).concat(Object.keys(pkg.devDependencies ?? {})) if (deps.includes('@arkstack/inertia')) { TSConfig['include'] = [...TSConfig['include'], '../resources'] @@ -36,7 +34,7 @@ export class BuildInterfaces { if (deps.includes('react')) { TSConfig['compilerOptions'] = { ...TSConfig['compilerOptions'], - jsx: 'react-jsx' + jsx: 'react-jsx', } as never } diff --git a/templates/express-inertia/resources/js/Pages/Index.tsx b/templates/express-inertia/resources/js/Pages/Index.tsx index de3d1f75..7ac71b4c 100644 --- a/templates/express-inertia/resources/js/Pages/Index.tsx +++ b/templates/express-inertia/resources/js/Pages/Index.tsx @@ -75,8 +75,7 @@ export default function Index({
Router.get('/', () => {'{'}
-   return {'{'} status:{' '} - 'ok' {'}'} +   return {'{'} status: 'ok' {'}'}
{'}'}) diff --git a/templates/express-inertia/src/app/http/controllers/BaseController.ts b/templates/express-inertia/src/app/http/controllers/BaseController.ts index 8355ad40..7588ba10 100644 --- a/templates/express-inertia/src/app/http/controllers/BaseController.ts +++ b/templates/express-inertia/src/app/http/controllers/BaseController.ts @@ -4,10 +4,7 @@ import { Controller } from 'clear-router' import { HttpContext } from 'clear-router/types/express' export class BaseController extends Controller { - async validate, R extends RulesForData>( - rules: R, - camelizeKeys = false, - ) { + async validate, R extends RulesForData>(rules: R, camelizeKeys = false) { const data = await Validator.make(this.body, rules).validate() if (camelizeKeys) { diff --git a/templates/express-inertia/src/config/filesystem.ts b/templates/express-inertia/src/config/filesystem.ts index bf209d4d..ac637d79 100644 --- a/templates/express-inertia/src/config/filesystem.ts +++ b/templates/express-inertia/src/config/filesystem.ts @@ -62,10 +62,7 @@ export default (): FilesystemConfig => { * The key of the object represents the link location, while the value represents the target location. */ links: { - [path.join(Arkstack.rootDir(), './public/storage')]: path.join( - Arkstack.rootDir(), - './storage/app/public', - ), + [path.join(Arkstack.rootDir(), './public/storage')]: path.join(Arkstack.rootDir(), './storage/app/public'), }, /** diff --git a/templates/express-inertia/tsdown.config.ts b/templates/express-inertia/tsdown.config.ts index dd003efa..b1965105 100644 --- a/templates/express-inertia/tsdown.config.ts +++ b/templates/express-inertia/tsdown.config.ts @@ -23,10 +23,7 @@ export default defineConfig([ deps: { skipNodeModulesBundle: true, }, - watch: - env === 'dev' && process.env.CLI_BUILD !== 'true' - ? ['.env', '.env.*', 'src', 'tsconfig.json'] - : false, + watch: env === 'dev' && process.env.CLI_BUILD !== 'true' ? ['.env', '.env.*', 'src', 'tsconfig.json'] : false, plugins: [ devServer({ env: Object.assign({}, process.env, { From 27f0bc8e6c8cec65cbbbe7a14be027106c16ebae Mon Sep 17 00:00:00 2001 From: 3m1n3nc3 Date: Wed, 21 Oct 2026 00:03:24 +0100 Subject: [PATCH 2/2] feat(queue): priority lists and * wildcard for queue:work `ark queue:work` could only work one queue, and a bare worker silently ignores jobs dispatched to any other queue. `--queue` now accepts: - a comma-separated priority list (`--queue=webhook,ocr,default`): each loop takes the first job found walking the list; - `*` for every queue on the connection, combinable with names to prioritize some (`--queue=webhook,*`). A bare `queue:work` still works only the default queue. `QueueContract.popAny()` backs the wildcard. The database driver takes the oldest due job across all queues (FIFO, so no queue starves); the redis driver discovers queues with SCAN (never KEYS) and works them round-robin. The base implementation falls back to the default queue, so drivers without it behave as before. --- CHANGELOG.md | 1 + docs/guide/jobs.md | 2 + docs/guide/queue.md | 23 +- packages/queue/README.md | 23 +- packages/queue/src/Contracts/QueueContract.ts | 10 + packages/queue/src/Worker.ts | 39 ++- .../queue/src/commands/QueueWorkCommand.ts | 36 ++- packages/queue/src/drivers/DatabaseQueue.ts | 48 +++- packages/queue/src/drivers/RedisQueue.ts | 45 +++ packages/queue/src/types.ts | 1 + packages/queue/tests/MemoryQueue.ts | 77 ++++-- packages/queue/tests/multi-queue.test.ts | 259 ++++++++++++++++++ 12 files changed, 513 insertions(+), 51 deletions(-) create mode 100644 packages/queue/tests/multi-queue.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 4d507bd5..cbb917be 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,7 @@ The format follows semantic versioning principles. ### Added +- `ark queue:work --queue` accepts a comma-separated priority list (`--queue=webhook,ocr,default`) and the `*` wildcard for every queue on the connection (`--queue=*`, or `--queue=webhook,*` to put some first). `Worker.runNextJob()` and `daemon({ queue })` take a string or a list. `QueueContract.popAny()` backs the wildcard: the `database` driver takes the oldest due job across all queues, and the `redis` driver discovers queues with `SCAN` and works them round-robin. A bare `queue:work` still works only the default queue. - `transports.pusher.public_host` / `public_port` / `public_use_tls` (`PUSHER_PUBLIC_*`) set the Pusher-protocol server address handed to browsers when it differs from the one the app uses, such as soketi inside Docker or behind a TLS proxy. Each defaults to its server-side counterpart. - `RealtimeConfig.importModule` loads the transport SDKs (`pusher-js`, `firebase/*`) for every transport and for device tokens, so pages without a bundler can load them from a CDN. Defaults to a dynamic `import()`. The `createPusherTransport`, `createFirebaseTransport`, `createFirebaseDatabaseTransport` and `requestDeviceToken` helpers accept it as an extra argument. - `RealtimeClient.deviceToken({ serviceWorker })` takes a service worker registration or script URL (registered with the Firebase web config as `firebaseConfig`), returns `null` where web push is unsupported, and falls back to the `firebase` config when streaming over another transport. `requestDeviceToken()` issues a token without a client. diff --git a/docs/guide/jobs.md b/docs/guide/jobs.md index 99e56652..edbe1865 100644 --- a/docs/guide/jobs.md +++ b/docs/guide/jobs.md @@ -87,6 +87,8 @@ With the default `sync` connection the job runs immediately. Configure a `databa ark queue:work ``` +A bare `ark queue:work` processes only the connection's default queue. If jobs go to other queues (`onQueue(...)` or a `queue` property), name them, e.g. `ark queue:work --queue=emails,default`, or use `ark queue:work --queue=*` to work them all. + ## Retries & failure handling Control attempts and backoff with instance properties, and react to permanent failure with a `failed` hook: diff --git a/docs/guide/queue.md b/docs/guide/queue.md index 11a5d1e7..c15d46e0 100644 --- a/docs/guide/queue.md +++ b/docs/guide/queue.md @@ -88,12 +88,29 @@ With the `sync` connection, pushed work runs immediately and inline. A worker pulls jobs from a connection and runs them. Run one from the CLI: ```sh -ark queue:work # daemon on the default connection +ark queue:work # daemon on the default connection's default queue ark queue:work redis --queue=emails +ark queue:work --queue=high,default # priority list: high first, then default +ark queue:work --queue=* # every queue on the connection ark queue:work --once # process a single job and exit ark queue:clear redis --queue=emails ``` +`--queue` takes one queue, a comma-separated list, or `*`: + +- **No `--queue`**: only the connection's default queue (`queue` in its config). Jobs + sent elsewhere with `onQueue(...)` or a job's `queue` property are **not** processed. +- **A list** (`--queue=webhook,ocr,default`) is a strict priority order: on every + loop the worker takes the first job it finds walking the list, so later queues + run only while earlier ones are empty. +- **`*`** works every queue on the connection. The `database` driver takes the + oldest due job across all queues; the `redis` driver discovers queues with + `SCAN` and works them round-robin. Combine it with names to prioritize some: + `--queue=webhook,*`. + +Every queue that receives jobs needs a worker that lists it (or `*`), or its jobs +wait forever. + Or drive a worker programmatically: ```ts @@ -101,8 +118,12 @@ import { Queue } from '@arkstack/queue' const worker = Queue.worker('database') await worker.daemon({ queue: 'default', sleep: 3 }) +await worker.daemon({ queue: ['webhook', '*'], sleep: 3 }) // priority list / wildcard ``` +A custom driver supports `*` by overriding `popAny()` on `QueueContract`; the base +implementation works only the default queue. + ### Retries & failures A job is **deleted** on success. On failure it is **released** for another attempt until it exhausts its `tries`, at which point it is **marked failed** — invoking the job's `failed` hook — and removed. Released jobs become available again after the job's `backoff` (seconds). diff --git a/packages/queue/README.md b/packages/queue/README.md index 3be20dcd..31f904a5 100644 --- a/packages/queue/README.md +++ b/packages/queue/README.md @@ -49,12 +49,29 @@ await Queue.clear('emails') ## Workers ```bash -ark queue:work # daemon on the default connection +ark queue:work # daemon on the default connection's default queue ark queue:work redis --queue=emails +ark queue:work --queue=high,default # priority list: high first, then default +ark queue:work --queue=* # every queue on the connection ark queue:work --once # process a single job and exit ark queue:clear redis --queue=emails ``` +`--queue` takes one queue, a comma-separated list, or `*`: + +- **No `--queue`**: only the connection's default queue (`queue` in its config). Jobs + sent elsewhere with `onQueue(...)` or a job's `queue` property are **not** processed. +- **A list** (`--queue=webhook,ocr,default`) is a strict priority order: on every + loop the worker takes the first job it finds walking the list, so later queues + run only while earlier ones are empty. +- **`*`** works every queue on the connection. The `database` driver takes the + oldest due job across all queues; the `redis` driver discovers queues with + `SCAN` and works them round-robin. Combine it with names to prioritize some: + `--queue=webhook,*`. + +Every queue that receives jobs needs a worker that lists it (or `*`), or its jobs +wait forever. + Programmatically: ```ts @@ -62,8 +79,12 @@ import { Queue } from '@arkstack/queue' const worker = Queue.worker('database') await worker.daemon({ queue: 'default', sleep: 3 }) +await worker.daemon({ queue: ['webhook', '*'], sleep: 3 }) // priority list / wildcard ``` +A custom driver supports `*` by overriding `popAny()` on `QueueContract`; the base +implementation works only the default queue. + A job is **deleted** on success, **released** for another attempt on failure, and **marked failed** (invoking its `failed` hook) once it exhausts `tries`. ## Job (de)serialization diff --git a/packages/queue/src/Contracts/QueueContract.ts b/packages/queue/src/Contracts/QueueContract.ts index 9cb61dff..da36d275 100644 --- a/packages/queue/src/Contracts/QueueContract.ts +++ b/packages/queue/src/Contracts/QueueContract.ts @@ -51,6 +51,16 @@ export abstract class QueueContract { */ abstract pop(queue?: string): Promise + /** + * Reserve the next job from any queue on this connection, for the worker's `*` + * wildcard. Drivers that can list their queues override this; the default + * works only the default queue, so a driver without that ability behaves + * exactly as a worker given no queue. + */ + async popAny(): Promise { + return this.pop() + } + /** * The number of jobs waiting on the given queue. */ diff --git a/packages/queue/src/Worker.ts b/packages/queue/src/Worker.ts index baeec8fd..aee15a04 100644 --- a/packages/queue/src/Worker.ts +++ b/packages/queue/src/Worker.ts @@ -2,9 +2,16 @@ import { Job } from './Job' import { QueueContract } from './Contracts/QueueContract' import { resolveJob } from './serialization' +/** Matches every queue on the connection in a worker's queue list. */ +export const ANY_QUEUE = '*' + export interface WorkerOptions { - /** The queue name to pull from. */ - queue?: string + /** + * The queue(s) to pull from. A list is a strict priority order: each loop + * takes the first job found walking the list. {@link ANY_QUEUE} (`*`) matches + * any queue, so `['webhook', '*']` means "webhook first, then anything". + */ + queue?: string | string[] /** Seconds to wait when the queue is empty before polling again. */ sleep?: number /** Stop after processing this many jobs (0 = unlimited). */ @@ -64,10 +71,13 @@ export class Worker { /** * Pop and process the next job. Returns `true` if a job was processed, - * `false` when the queue was empty. + * `false` when every queue asked for was empty. + * + * @param queue A queue, or a priority-ordered list of queues (see + * {@link WorkerOptions.queue}). Omitted: the connection's default. */ - async runNextJob(queue?: string): Promise { - const job = await this.connection.pop(queue) + async runNextJob(queue?: string | string[]): Promise { + const job = await this.nextJob(queue) if (!job) { return false @@ -78,6 +88,25 @@ export class Worker { return true } + /** + * Reserve the next job, walking the queues in priority order. + * + * @param queue + */ + private async nextJob(queue?: string | string[]): Promise { + const queues = Array.isArray(queue) ? queue : [queue] + + for (const name of queues) { + const job = name === ANY_QUEUE ? await this.connection.popAny() : await this.connection.pop(name) + + if (job) { + return job + } + } + + return null + } + /** * Execute a single reserved job, handling success and failure. */ diff --git a/packages/queue/src/commands/QueueWorkCommand.ts b/packages/queue/src/commands/QueueWorkCommand.ts index 58e50b70..47d9baa6 100644 --- a/packages/queue/src/commands/QueueWorkCommand.ts +++ b/packages/queue/src/commands/QueueWorkCommand.ts @@ -1,6 +1,6 @@ import { Command } from '@h3ravel/musket' import { Queue } from '../QueueManager' -import { Worker } from '../Worker' +import { ANY_QUEUE, Worker } from '../Worker' import { bootArkorm } from '@arkstack/database' /** The message of a thrown value, whatever it is. */ @@ -12,7 +12,7 @@ const reason = (error: unknown): string => (error instanceof Error ? error.messa export class QueueWorkCommand extends Command { protected signature = `queue:work {connection? : The queue connection to work. Defaults to the configured default.} - {--queue= : The queue to process.} + {--queue= : The queue(s) to process: a name, a comma-separated priority list, or * for every queue.} {--sleep=3 : Seconds to sleep when no job is available.} {--max-jobs=0 : Stop after processing this many jobs (0 for unlimited).} {--once : Process a single job and exit.} @@ -35,7 +35,7 @@ export class QueueWorkCommand extends Command { }, }) - const queue = (this.option('queue') as string | undefined) ?? connection.getDefaultQueue() + const queue = this.queues(connection.getDefaultQueue()) try { bootArkorm() @@ -56,7 +56,7 @@ export class QueueWorkCommand extends Command { return } - this.info(`Processing jobs from the [${connection.getConnectionName()}] connection on the [${queue}] queue.`) + this.info(`Processing jobs from the [${connection.getConnectionName()}] connection on ${this.describe(queue)}.`) await worker.daemon({ queue, @@ -66,6 +66,34 @@ export class QueueWorkCommand extends Command { }) } + /** + * The queues to work, from `--queue`: `ocr` → `['ocr']`, `webhook,ocr` → a + * priority list, `*` → every queue. Falls back to the connection's default. + * + * @param fallback The connection's default queue. + */ + private queues(fallback: string): string[] { + const names = String(this.option('queue') ?? '') + .split(',') + .map((name) => name.trim()) + .filter(Boolean) + + return names.length ? [...new Set(names)] : [fallback] + } + + /** + * How the worker's queues read in its startup line. + * + * @param queues + */ + private describe(queues: string[]): string { + if (queues.length === 1) { + return queues[0] === ANY_QUEUE ? 'every queue' : `the [${queues[0]}] queue` + } + + return `the [${queues.join(', ')}] queues, in that order` + } + /** * Read a multi-word flag. Musket hands over the parsed options camelCased, * so `--stop-when-empty` arrives as `stopWhenEmpty`; the kebab-case name is diff --git a/packages/queue/src/drivers/DatabaseQueue.ts b/packages/queue/src/drivers/DatabaseQueue.ts index e8db159d..49624fc8 100644 --- a/packages/queue/src/drivers/DatabaseQueue.ts +++ b/packages/queue/src/drivers/DatabaseQueue.ts @@ -1,4 +1,4 @@ -import type { DatabaseConnectionConfig, DatabaseFacade, JobPayload, Query, Queueable } from '../types' +import type { DatabaseConnectionConfig, DatabaseFacade, JobPayload, JobRow, Query, Queueable } from '../types' import { Job } from '../Job' import { QueueContract } from '../Contracts/QueueContract' @@ -74,22 +74,40 @@ export class DatabaseQueue extends QueueContract { async pop(queue?: string): Promise { const db = await this.database() - const name = queue ?? this.defaultQueue // Next available, unreserved (or expired-reservation) job in id order. - const row = await this.availableQuery(db, name).orderBy({ id: 'asc' }).first() + const row = await this.availableQuery(db, queue ?? this.defaultQueue) + .orderBy({ id: 'asc' }) + .first() - if (!row) { - return null - } + return row ? this.reserve(db, row) : null + } + + /** + * The oldest due job on any queue. Ordering by id across queues is FIFO by + * arrival, so a busy queue can't starve a quiet one. + */ + async popAny(): Promise { + const db = await this.database() + const row = await this.availableQuery(db).orderBy({ id: 'asc' }).first() + + return row ? this.reserve(db, row) : null + } + /** + * Reserve a fetched row and wrap it as a {@link Job}. + * + * @param db + * @param row + */ + private async reserve(db: DatabaseFacade, row: JobRow): Promise { const attempts = row.attempts + 1 await db.table(this.options.table).where({ id: row.id }).update({ reserved_at: now(), attempts }) const payload = { ...(JSON.parse(row.payload) as JobPayload), attempts } - return new Job(payload, name, { + return new Job(payload, row.queue, { delete: async () => { await db.table(this.options.table).where({ id: row.id }).delete() }, @@ -125,18 +143,18 @@ export class DatabaseQueue extends QueueContract { } /** - * Due, unreserved jobs on the given queue, oldest first. Released jobs become - * unreserved with a future `available_at`, so they naturally wait their turn. + * Due, unreserved jobs, oldest first: on one queue, or on every queue when + * `queue` is omitted. Released jobs become unreserved with a future + * `available_at`, so they naturally wait their turn. * * @param db * @param queue * @returns */ - private availableQuery(db: DatabaseFacade, queue: string): Query { - return db - .table(this.options.table) - .where({ queue }) - .whereNull('reserved_at') - .whereRaw('available_at <= ' + now()) + private availableQuery(db: DatabaseFacade, queue?: string): Query { + const table = db.table(this.options.table) + const scoped = queue === undefined ? table : table.where({ queue }) + + return scoped.whereNull('reserved_at').whereRaw('available_at <= ' + now()) } } diff --git a/packages/queue/src/drivers/RedisQueue.ts b/packages/queue/src/drivers/RedisQueue.ts index eda47a3a..1d97ecf9 100644 --- a/packages/queue/src/drivers/RedisQueue.ts +++ b/packages/queue/src/drivers/RedisQueue.ts @@ -16,6 +16,8 @@ const now = (): number => Math.floor(Date.now() / 1000) */ export class RedisQueue extends QueueContract { private client?: RedisClient + /** Rotates which queue {@link popAny} tries first, so none is starved. */ + private anyCursor = 0 constructor(private readonly options: RedisConnectionConfig) { super() @@ -126,6 +128,27 @@ export class RedisQueue extends QueueContract { }) } + /** + * Reserve the next job from any queue. Redis has no ordering across lists, so + * the queues are discovered with `SCAN` (never the blocking `KEYS`) and tried + * round-robin, starting one further along on each call. + */ + async popAny(): Promise { + const queues = await this.queueNames(await this.connection()) + + for (let offset = 0; offset < queues.length; offset++) { + const job = await this.pop(queues[(this.anyCursor + offset) % queues.length]) + + if (job) { + this.anyCursor = (this.anyCursor + offset + 1) % queues.length + + return job + } + } + + return null + } + async size(queue?: string): Promise { const client = await this.connection() const name = queue ?? this.defaultQueue @@ -149,6 +172,28 @@ export class RedisQueue extends QueueContract { return count } + /** + * Every queue with jobs in any state (waiting, delayed or reserved), sorted so + * the round-robin order is stable. + */ + private async queueNames(client: RedisClient): Promise { + const prefix = this.options.prefix ?? 'queues:' + const names = new Set() + let cursor = '0' + + do { + const [next, keys] = await client.scan(cursor, 'MATCH', `${prefix}*`, 'COUNT', 100) + + for (const key of keys) { + names.add(key.slice(prefix.length).replace(/:(delayed|reserved)$/, '')) + } + + cursor = next + } while (cursor !== '0') + + return [...names].sort() + } + /** * Move due delayed jobs and expired reservations back onto the waiting list. */ diff --git a/packages/queue/src/types.ts b/packages/queue/src/types.ts index 3cac612e..b309854c 100644 --- a/packages/queue/src/types.ts +++ b/packages/queue/src/types.ts @@ -152,5 +152,6 @@ export interface RedisClient { zrem(key: string, member: string): Promise zcard(key: string): Promise zrangebyscore(key: string, min: string | number, max: string | number): Promise + scan(cursor: string | number, ...args: (string | number)[]): Promise<[cursor: string, keys: string[]]> quit(): Promise } diff --git a/packages/queue/tests/MemoryQueue.ts b/packages/queue/tests/MemoryQueue.ts index fe1a04de..4716af99 100644 --- a/packages/queue/tests/MemoryQueue.ts +++ b/packages/queue/tests/MemoryQueue.ts @@ -2,56 +2,83 @@ import { Job, type JobPayload, QueueContract, type Queueable, serializeJob } fro /** * A simple in-memory queue connection used to exercise the worker, reservation, - * release and failure flows without needing Redis or a database. + * release and failure flows without needing Redis or a database. Jobs are kept + * per named queue (unnamed pushes and pops use `default`), and a global + * sequence gives {@link popAny} the same oldest-first order the database driver + * has. */ export class MemoryQueue extends QueueContract { - private jobs: JobPayload[] = [] + private queues = new Map>() private reserved = new Map() + private seq = 0 - async push(job: Queueable, _queue?: string): Promise { - return this.pushRaw(serializeJob(job)) + async push(job: Queueable, queue?: string): Promise { + return this.pushRaw(serializeJob(job), queue ?? job.queue) } - async pushRaw(payload: JobPayload): Promise { - this.jobs.push(payload) + async pushRaw(payload: JobPayload, queue?: string): Promise { + const name = queue ?? this.getDefaultQueue() + const list = this.queues.get(name) ?? [] + + list.push({ seq: this.seq++, payload }) + this.queues.set(name, list) return payload.id } - async later(_delay: number | Date, job: Queueable): Promise { - return this.push(job) + async later(_delay: number | Date, job: Queueable, queue?: string): Promise { + return this.push(job, queue) } async pop(queue?: string): Promise { - const payload = this.jobs.shift() + const name = queue ?? this.getDefaultQueue() + const entry = this.queues.get(name)?.shift() + + return entry ? this.reserve(entry.payload, name) : null + } + + async popAny(): Promise { + let oldest: { name: string; seq: number } | null = null + + for (const [name, list] of this.queues) { + if (list.length && (!oldest || list[0].seq < oldest.seq)) { + oldest = { name, seq: list[0].seq } + } + } + + return oldest ? this.pop(oldest.name) : null + } - if (!payload) { - return null + async size(queue?: string): Promise { + if (queue !== undefined) { + return this.queues.get(queue)?.length ?? 0 } + const waiting = [...this.queues.values()].reduce((total, list) => total + list.length, 0) + + return waiting + this.reserved.size + } + + async clear(): Promise { + const count = await this.size() + this.queues.clear() + this.reserved.clear() + + return count + } + + private reserve(payload: JobPayload, queue: string): Job { payload.attempts += 1 this.reserved.set(payload.id, payload) - return new Job(payload, queue ?? 'default', { + return new Job(payload, queue, { delete: async () => { this.reserved.delete(payload.id) }, release: async () => { this.reserved.delete(payload.id) - this.jobs.push(payload) + await this.pushRaw(payload, queue) }, }) } - - async size(): Promise { - return this.jobs.length + this.reserved.size - } - - async clear(): Promise { - const count = await this.size() - this.jobs = [] - this.reserved.clear() - - return count - } } diff --git a/packages/queue/tests/multi-queue.test.ts b/packages/queue/tests/multi-queue.test.ts new file mode 100644 index 00000000..67ad5baa --- /dev/null +++ b/packages/queue/tests/multi-queue.test.ts @@ -0,0 +1,259 @@ +import type { DatabaseFacade, JobRow, Query, RedisClient } from '../src' +import { DatabaseQueue, Queue, RedisQueue, Worker, resetSerialization, serializeJob } from '../src' +import { afterEach, beforeEach, describe, expect, it } from 'vitest' + +import { MemoryQueue } from './MemoryQueue' + +/** A job that records which queue's work ran, in order. */ +class TaggedJob { + static log: string[] = [] + + constructor( + public tag: string, + public queue?: string, + ) {} + + handle() { + TaggedJob.log.push(this.tag) + } + + serialize() { + return { tag: this.tag } + } +} + +const job = (tag: string, queue?: string) => new TaggedJob(tag, queue) as never + +describe('Multi-queue workers', () => { + beforeEach(() => { + Queue.resolveJobsUsing((payload) => new TaggedJob(payload.data.tag as string)) + }) + + afterEach(() => { + resetSerialization() + TaggedJob.log = [] + }) + + describe('priority list', () => { + it('takes the first job walking the list, whatever order jobs arrived in', async () => { + const queue = new MemoryQueue() + await queue.push(job('low-1', 'low')) + await queue.push(job('high-1', 'high')) + await queue.push(job('low-2', 'low')) + await queue.push(job('high-2', 'high')) + + await new Worker(queue).daemon({ queue: ['high', 'low'], sleep: 0, stopWhenEmpty: true }) + + expect(TaggedJob.log).toEqual(['high-1', 'high-2', 'low-1', 'low-2']) + }) + + it('skips empty queues and reports empty only when every listed queue is', async () => { + const queue = new MemoryQueue() + await queue.push(job('second', 'b')) + const worker = new Worker(queue) + + expect(await worker.runNextJob(['a', 'b'])).toBe(true) + expect(await worker.runNextJob(['a', 'b'])).toBe(false) + expect(TaggedJob.log).toEqual(['second']) + }) + + it('never touches a queue that is not listed', async () => { + const queue = new MemoryQueue() + await queue.push(job('elsewhere', 'other')) + + expect(await new Worker(queue).runNextJob(['a', 'b'])).toBe(false) + expect(await queue.size('other')).toBe(1) + }) + }) + + describe('* wildcard', () => { + it('drains every queue, oldest job first', async () => { + const queue = new MemoryQueue() + await queue.push(job('ocr-1', 'ocr')) + await queue.push(job('webhook-1', 'webhook')) + await queue.push(job('default-1')) + await queue.push(job('ocr-2', 'ocr')) + + await new Worker(queue).daemon({ queue: ['*'], sleep: 0, stopWhenEmpty: true }) + + expect(TaggedJob.log).toEqual(['ocr-1', 'webhook-1', 'default-1', 'ocr-2']) + expect(await queue.size()).toBe(0) + }) + + it('combines with named queues: listed queues first, then anything', async () => { + const queue = new MemoryQueue() + await queue.push(job('ocr', 'ocr')) + await queue.push(job('webhook', 'webhook')) + + await new Worker(queue).daemon({ queue: ['webhook', '*'], sleep: 0, stopWhenEmpty: true }) + + expect(TaggedJob.log).toEqual(['webhook', 'ocr']) + }) + }) + + describe('DatabaseQueue.popAny', () => { + /** A fake table that applies the where/whereNull filters `availableQuery` builds. */ + const fakeDatabase = (rows: JobRow[]) => { + const queries: Array> = [] + + const query = (filters: Record, matching: () => JobRow[]): Query => + ({ + where: (where: Record) => { + const next = { ...filters, ...where } + + return query(next, () => + matching().filter((row) => + Object.entries(where).every(([key, value]) => row[key as keyof JobRow] === value), + ), + ) + }, + whereNull: (column: string) => + query(filters, () => matching().filter((row) => row[column as keyof JobRow] === null)), + whereRaw: () => query(filters, matching), + orderBy: () => query(filters, () => [...matching()].sort((a, b) => Number(a.id) - Number(b.id))), + first: async () => { + queries.push(filters) + + return matching()[0] ?? null + }, + update: async (values: Record) => { + for (const row of matching()) Object.assign(row, values) + }, + delete: async () => {}, + count: async () => matching().length, + }) as Query + + const db = { table: () => query({}, () => rows) } as unknown as DatabaseFacade + + return { db, queries } + } + + const row = (id: number, queue: string, tag: string): JobRow => ({ + id, + queue, + payload: JSON.stringify(serializeJob(job(tag))), + attempts: 0, + reserved_at: null, + available_at: 0, + created_at: 0, + }) + + it('reserves the oldest due job on any queue, tagged with its own queue', async () => { + const { db, queries } = fakeDatabase([row(2, 'webhook', 'b'), row(1, 'ocr', 'a')]) + const queue = new DatabaseQueue({ driver: 'database', table: 'jobs' }) + ;(queue as unknown as { db: DatabaseFacade }).db = db + + const reserved = await queue.popAny() + + expect(reserved?.queue).toBe('ocr') + expect(reserved?.payload.data.tag).toBe('a') + // No queue filter: the wildcard spans every queue. + expect(queries.at(-1)).not.toHaveProperty('queue') + }) + + it('still scopes a named pop to its queue', async () => { + const { db } = fakeDatabase([row(1, 'ocr', 'a'), row(2, 'webhook', 'b')]) + const queue = new DatabaseQueue({ driver: 'database', table: 'jobs' }) + ;(queue as unknown as { db: DatabaseFacade }).db = db + + expect((await queue.pop('webhook'))?.payload.data.tag).toBe('b') + expect(await queue.pop('missing')).toBeNull() + }) + }) + + describe('RedisQueue.popAny', () => { + /** Just enough of ioredis (lists, sorted sets, SCAN) for the queue driver. */ + const fakeRedis = () => { + const lists = new Map() + const zsets = new Map>() + const scans: string[] = [] + + const client: RedisClient = { + rpush: async (key, ...values) => { + const list = lists.get(key) ?? [] + list.push(...values) + lists.set(key, list) + + return list.length + }, + lpop: async (key) => lists.get(key)?.shift() ?? null, + llen: async (key) => lists.get(key)?.length ?? 0, + del: async (...keys) => keys.filter((key) => lists.delete(key) || zsets.delete(key)).length, + zadd: async (key, score, member) => { + const set = zsets.get(key) ?? new Map() + set.set(member, score) + zsets.set(key, set) + + return 1 + }, + zrem: async (key, member) => Number(zsets.get(key)?.delete(member) ?? false), + zcard: async (key) => zsets.get(key)?.size ?? 0, + zrangebyscore: async (key, _min, max) => { + const top = max === '+inf' ? Infinity : Number(max) + + return [...(zsets.get(key) ?? [])].filter(([, score]) => score <= top).map(([member]) => member) + }, + // Returns the matching keys over two pages, to exercise the cursor loop. + scan: async (cursor, ..._args) => { + scans.push(String(cursor)) + const keys = [...lists.keys(), ...zsets.keys()].filter( + (key) => (lists.get(key)?.length ?? 0) > 0 || (zsets.get(key)?.size ?? 0) > 0, + ) + const half = Math.ceil(keys.length / 2) + + return cursor === '0' ? ['1', keys.slice(0, half)] : ['0', keys.slice(half)] + }, + quit: async () => 'OK', + } + + return { client, scans } + } + + const redisQueue = (client: RedisClient) => { + const queue = new RedisQueue({ driver: 'redis' }) + ;(queue as unknown as { client: RedisClient }).client = client + + return queue + } + + it('discovers queues with SCAN and works them round-robin', async () => { + const { client, scans } = fakeRedis() + const queue = redisQueue(client) + await queue.push(job('a1', 'alpha')) + await queue.push(job('a2', 'alpha')) + await queue.push(job('b1', 'beta')) + await queue.push(job('b2', 'beta')) + + const order: string[] = [] + for (let i = 0; i < 4; i++) { + const reserved = await queue.popAny() + order.push(`${reserved?.queue}:${reserved?.payload.data.tag}`) + await reserved?.delete() + } + + // Alternates rather than draining `alpha` before `beta` gets a turn. + expect(order).toEqual(['alpha:a1', 'beta:b1', 'alpha:a2', 'beta:b2']) + expect(await queue.popAny()).toBeNull() + expect(scans).toContain('1') + }) + + it('finds a queue that only has delayed jobs once they are due', async () => { + const { client } = fakeRedis() + const queue = redisQueue(client) + await queue.pushRaw(serializeJob(job('later')), 'scheduled', Math.floor(Date.now() / 1000) + 3600) + + expect(await queue.popAny()).toBeNull() + + // Make it due, as time passing would. + await client.zadd( + 'queues:scheduled:delayed', + 0, + (await client.zrangebyscore('queues:scheduled:delayed', '-inf', '+inf'))[0], + ) + const reserved = await queue.popAny() + + expect(reserved?.queue).toBe('scheduled') + expect(reserved?.payload.data.tag).toBe('later') + }) + }) +})