diff --git a/docs/guide/jobs.md b/docs/guide/jobs.md index 0ab63d47..4dcccbac 100644 --- a/docs/guide/jobs.md +++ b/docs/guide/jobs.md @@ -122,4 +122,12 @@ serialize() { } ``` -For dedicated worker **processes**, make sure your job modules are imported — or call `JobRegistry.register(MyJob)` — so the names are known before jobs are processed. +A dedicated worker **process** constructs none of your jobs, so `ark queue:work` imports every job module in `src/app/jobs` (or its build output) before it starts working, registering each class it finds under its class name. Jobs kept elsewhere can be registered by loading that directory instead: + +```ts +import { loadJobs } from '@arkstack/jobs'; + +await loadJobs('domain/jobs'); // relative to src/, or the build output +``` + +`JobRegistry.register(MyJob)` still registers a single class explicitly. diff --git a/packages/jobs/src/JobRegistry.ts b/packages/jobs/src/JobRegistry.ts index ca5a75db..6695f3c0 100644 --- a/packages/jobs/src/JobRegistry.ts +++ b/packages/jobs/src/JobRegistry.ts @@ -2,17 +2,30 @@ import type { JobPayload } from '@arkstack/queue' import type { Job } from './Job' import type { JobConstructor } from './types' +// Back the registry with a global symbol so a job module evaluated more than +// once — the app's own import and a worker's jiti-loaded copy are different +// module instances — registers into the one map the worker resolves against. +const REGISTRY = Symbol.for('arkstack.jobs.registry') + +const registry = (): Map => { + const store = globalThis as unknown as Record> + + return (store[REGISTRY] ??= new Map()) +} + /** * A registry mapping job names to their classes so a worker can reconstruct job * instances from a stored payload. * * Job classes register themselves when instantiated, which covers same-process - * dispatch + work. For dedicated worker processes, ensure the job modules are - * imported (or call {@link JobRegistry.register} explicitly) so the names are - * known before jobs are processed. + * dispatch + work. A dedicated worker process constructs none of them, so + * `queue:work` calls {@link loadJobs} to import the application's job modules + * before it starts working. */ export class JobRegistry { - private static classes = new Map() + private static get classes(): Map { + return registry() + } /** * Register a job class under an explicit name (defaults to the class name). diff --git a/packages/jobs/src/index.ts b/packages/jobs/src/index.ts index d20917b6..2082607a 100644 --- a/packages/jobs/src/index.ts +++ b/packages/jobs/src/index.ts @@ -7,5 +7,6 @@ export * from './Job' export * from './JobRegistry' export * from './PendingDispatch' export * from './dispatch' +export * from './loader' export * from './bridge' export * from './types' diff --git a/packages/jobs/src/loader.ts b/packages/jobs/src/loader.ts new file mode 100644 index 00000000..7da8b971 --- /dev/null +++ b/packages/jobs/src/loader.ts @@ -0,0 +1,80 @@ +import { importFile, nodeEnv, outputDir } from '@arkstack/common' +import path, { join } from 'node:path' + +import { Arkstack } from '@arkstack/contract' +import { Dirent } from 'node:fs' +import { JobConstructor } from './types' +import { JobRegistry } from './JobRegistry' +import { readdirSync } from 'node:fs' + +/** + * Whether a module export is a runnable job class. + * + * The test is `handle` on the prototype rather than `instanceof Job`: a job + * module loaded through jiti carries its own copy of the base class, so an + * identity check would reject every job it finds. An abstract base declares no + * `handle` of its own and is skipped. + */ +const isJobClass = (value: unknown): value is JobConstructor => typeof value === 'function' + && typeof (value as { prototype?: { handle?: unknown } }).prototype?.handle === 'function' + +/** The job modules in the first candidate directory that has any. */ +const jobFiles = (directories: string[]): [string, Dirent[]] => { + for (const directory of directories) { + try { + const files = readdirSync(directory, { withFileTypes: true }).filter( + (file) => file.isFile() && ['.ts', '.js', '.mjs'].includes(path.extname(file.name)), + ) + + if (files.length) return [directory, files] + } catch { + // Directory missing — try the next candidate. + } + } + + return ['', []] +} + +/** + * Import the application's job classes so a worker can reconstruct them. + * + * A dedicated `queue:work` process never constructs the app's jobs itself, and + * a job class only reaches the {@link JobRegistry} when its module is loaded — + * without this a worker cannot resolve a single payload it pops, and every job + * is released back onto the queue until it exhausts its attempts. + * + * Every concrete `Job` subclass a module exports is registered under its class + * name, which is the name payloads are serialized with. + * + * @param subPath Directory (under `src/`, or the build output) to load from. + * @returns The names registered, in discovery order. + */ +export const loadJobs = async (subPath: string = join('app', 'jobs')): Promise => { + const root = Arkstack.rootDir() + + // Production prefers the build output (a deploy ships only `dist`); dev + // prefers source so edits land without a rebuild. Either falls back. + const [directory, files] = jobFiles(nodeEnv() === 'prod' + ? [join(outputDir(), subPath), join(root, 'src', subPath)] + : [join(root, 'src', subPath), join(outputDir(), subPath)]) + + const registered: string[] = [] + + for (const file of files) { + // One unloadable job module must not cost the worker every other job. + try { + const module = await importFile>(join(directory, file.name)) + + for (const value of Object.values(module)) { + if (!isJobClass(value)) continue + + JobRegistry.register(value) + registered.push(value.name) + } + } catch { + continue + } + } + + return registered +} diff --git a/packages/jobs/tests/jobs.test.ts b/packages/jobs/tests/jobs.test.ts index f34e0e26..440dea3b 100644 --- a/packages/jobs/tests/jobs.test.ts +++ b/packages/jobs/tests/jobs.test.ts @@ -1,4 +1,4 @@ -import { Job, JobRegistry, dispatch } from '../src' +import { Job, JobRegistry, dispatch, loadJobs } from '../src' import { type JobPayload, QueueContract, type Queueable, Worker, serializeJob } from '@arkstack/queue' import { afterAll, afterEach, beforeAll, describe, expect, it } from 'vitest' import { dirname, resolve } from 'node:path' @@ -153,6 +153,36 @@ describe('Jobs', () => { }) }) + describe('loadJobs', () => { + it('registers the job classes a worker process would otherwise not know', async () => { + // A dedicated worker constructs none of the app's jobs, so nothing + // reaches the registry until their modules are loaded. + JobRegistry.clear() + + expect(JobRegistry.has('LoadedJob')).toBe(false) + + const registered = await loadJobs() + + expect(registered).toContain('LoadedJob') + expect(registered).toContain('AlsoLoadedJob') + expect(JobRegistry.has('LoadedJob')).toBe(true) + expect(JobRegistry.has('AlsoLoadedJob')).toBe(true) + }) + + it('skips abstract bases, non-job exports and modules that fail to load', async () => { + const registered = await loadJobs() + + expect(registered).not.toContain('BaseFixtureJob') + expect(registered).not.toContain('notAJob') + // BrokenJob.ts throws on import; the rest still load. + expect(registered).toContain('LoadedJob') + }) + + it('returns nothing when the directory does not exist', async () => { + expect(await loadJobs('app/nowhere')).toEqual([]) + }) + }) + describe('End to end with a worker', () => { it('serializes, stores, resolves and runs the job', async () => { const queue = new MemoryQueue() diff --git a/packages/jobs/tests/src/app/jobs/BrokenJob.ts b/packages/jobs/tests/src/app/jobs/BrokenJob.ts new file mode 100644 index 00000000..2317c668 --- /dev/null +++ b/packages/jobs/tests/src/app/jobs/BrokenJob.ts @@ -0,0 +1 @@ +throw new Error('this module cannot be loaded') diff --git a/packages/jobs/tests/src/app/jobs/FixtureJobs.ts b/packages/jobs/tests/src/app/jobs/FixtureJobs.ts new file mode 100644 index 00000000..d6b4b6f0 --- /dev/null +++ b/packages/jobs/tests/src/app/jobs/FixtureJobs.ts @@ -0,0 +1,15 @@ +import { Job } from '../../../../src' + +/** An abstract base: not instantiable, so it must not be registered. */ +export abstract class BaseFixtureJob extends Job { } + +export class LoadedJob extends BaseFixtureJob { + async handle () { } +} + +export class AlsoLoadedJob extends Job { + async handle () { } +} + +/** A plain export that isn't a job at all. */ +export const notAJob = () => 'nope' diff --git a/packages/queue/src/Contracts/QueueContract.ts b/packages/queue/src/Contracts/QueueContract.ts index 6fe8045f..968a2316 100644 --- a/packages/queue/src/Contracts/QueueContract.ts +++ b/packages/queue/src/Contracts/QueueContract.ts @@ -23,6 +23,14 @@ export abstract class QueueContract { return this } + /** + * The queue this connection works when none is named. Drivers with a + * configurable queue override it. + */ + getDefaultQueue (): string { + return 'default' + } + /** * Push a job onto the queue. Returns the job id. */ diff --git a/packages/queue/src/Worker.ts b/packages/queue/src/Worker.ts index 90563586..57e90955 100644 --- a/packages/queue/src/Worker.ts +++ b/packages/queue/src/Worker.ts @@ -13,6 +13,23 @@ export interface WorkerOptions { stopWhenEmpty?: boolean } +/** + * Observers for what a worker does with each job. A worker executes jobs the + * caller can't see — a dedicated worker process reports what happened through + * these, so a job that keeps failing isn't indistinguishable from an idle queue. + */ +export interface WorkerHandlers { + /** A job has been reserved and is about to run. */ + onProcessing?: (job: Job) => void | Promise + /** A job completed and has been removed from the queue. */ + onProcessed?: (job: Job) => void | Promise + /** + * A job threw. `released` is `true` when it goes back on the queue for + * another attempt, `false` when it has exhausted its tries and failed. + */ + onFailed?: (job: Job, error: unknown, released: boolean) => void | Promise +} + const wait = (ms: number): Promise => new Promise((resolve) => setTimeout(resolve, ms)) /** @@ -24,6 +41,7 @@ const wait = (ms: number): Promise => new Promise((resolve) => setTimeout( */ export class Worker { private shouldStop = false + private readonly handlers: WorkerHandlers = {} constructor(private readonly connection: QueueContract) { } @@ -32,6 +50,18 @@ export class Worker { this.shouldStop = true } + /** + * Observe what the worker does with each job. Handlers merge, so they can be + * registered in more than one call. + * + * @param handlers The callbacks to add. + */ + on (handlers: WorkerHandlers): this { + Object.assign(this.handlers, handlers) + + return this + } + /** * Pop and process the next job. Returns `true` if a job was processed, * `false` when the queue was empty. @@ -53,10 +83,13 @@ export class Worker { */ async process (job: Job): Promise { try { + await this.handlers.onProcessing?.(job) + const instance = await resolveJob(job.payload) await instance.handle() await job.delete() + await this.handlers.onProcessed?.(job) } catch (error) { await this.handleFailure(job, error) } @@ -107,10 +140,12 @@ export class Worker { } await job.delete() + await this.handlers.onFailed?.(job, error, false) return } await job.release(job.backoff()) + await this.handlers.onFailed?.(job, error, true) } } diff --git a/packages/queue/src/commands/QueueWorkCommand.ts b/packages/queue/src/commands/QueueWorkCommand.ts index 2755b234..b68be8ba 100644 --- a/packages/queue/src/commands/QueueWorkCommand.ts +++ b/packages/queue/src/commands/QueueWorkCommand.ts @@ -1,7 +1,13 @@ import { Command } from '@h3ravel/musket' import { Queue } from '../QueueManager' +import { Worker } from '../Worker' import { bootArkorm } from '@arkstack/database' +/** The message of a thrown value, whatever it is. */ +const reason = (error: unknown): string => error instanceof Error + ? error.message + : String(error) + /** * Process jobs from a queue connection. */ @@ -17,29 +23,72 @@ export class QueueWorkCommand extends Command { protected description = 'Start processing jobs on the queue as a daemon.' async handle() { - const connection = this.argument('connection') as string | undefined - const worker = Queue.worker(connection) - const queue = this.option('queue') as string | undefined + const connection = Queue.connection(this.argument('connection') as string | undefined) + const worker = new Worker(connection).on({ + onProcessed: (job) => { + this.success(`Processed: ${job.name()}`) + }, + onFailed: (job, error, released) => { + this.error(released + ? `Failed (attempt ${job.attempts()}, retrying): ${job.name()} — ${reason(error)}` + : `Failed permanently: ${job.name()} — ${reason(error)}`) + }, + }) + + const queue = (this.option('queue') as string | undefined) ?? connection.getDefaultQueue() + + try { + bootArkorm() + } catch {/** */ } + + // The worker runs in its own process, so nothing has constructed the + // application's jobs and the registry a payload resolves against is + // empty. Load them before working, or every job pops and fails. + await this.registerJobs() if (this.option('once')) { const handled = await worker.runNextJob(queue) - this.info(handled ? 'Processed one job.' : 'No jobs available.') + if (!handled) this.info('No jobs available.') return } - this.info(`Processing jobs from [${connection ?? 'default'}] connection.`) - - try { - bootArkorm() - } catch {/** */ } + this.info(`Processing jobs from the [${connection.getConnectionName()}] connection on the [${queue}] queue.`) await worker.daemon({ queue, sleep: Number(this.option('sleep') ?? 3), - maxJobs: Number(this.option('max-jobs') ?? 0), - stopWhenEmpty: Boolean(this.option('stop-when-empty')), + maxJobs: Number(this.flag('maxJobs', 'max-jobs') ?? 0), + stopWhenEmpty: Boolean(this.flag('stopWhenEmpty', 'stop-when-empty')), }) } + + /** + * Read a multi-word flag. Musket hands over the parsed options camelCased, + * so `--stop-when-empty` arrives as `stopWhenEmpty`; the kebab-case name is + * accepted too so the flag can't go quietly missing again. + * + * @param camel The camelCase key. + * @param kebab The flag as written in the signature. + */ + private flag(camel: string, kebab: string): unknown { + return this.option(camel) ?? this.option(kebab) + } + + /** + * Register the application's job classes with `@arkstack/jobs` when it is + * installed. Without it a payload cannot be turned back into a job. + */ + private async registerJobs(): Promise { + try { + const specifier = '@arkstack/jobs' + const { loadJobs } = await import(specifier) + const names = (await loadJobs()) as string[] + + if (names.length) this.line(`Loaded ${names.length} job class(es).`) + } catch { + // `@arkstack/jobs` isn't installed; the app registers its own jobs. + } + } } diff --git a/packages/queue/src/drivers/DatabaseQueue.ts b/packages/queue/src/drivers/DatabaseQueue.ts index 30c94e9d..0ad77303 100644 --- a/packages/queue/src/drivers/DatabaseQueue.ts +++ b/packages/queue/src/drivers/DatabaseQueue.ts @@ -24,6 +24,11 @@ export class DatabaseQueue extends QueueContract { } private get defaultQueue(): string { + return this.getDefaultQueue() + } + + /** The configured queue this connection works when none is named. */ + getDefaultQueue(): string { return this.options.queue ?? 'default' } diff --git a/packages/queue/src/drivers/RedisQueue.ts b/packages/queue/src/drivers/RedisQueue.ts index 66907099..687881c9 100644 --- a/packages/queue/src/drivers/RedisQueue.ts +++ b/packages/queue/src/drivers/RedisQueue.ts @@ -22,6 +22,11 @@ export class RedisQueue extends QueueContract { } private get defaultQueue (): string { + return this.getDefaultQueue() + } + + /** The configured queue this connection works when none is named. */ + getDefaultQueue (): string { return this.options.queue ?? 'default' } diff --git a/packages/queue/tests/queue.test.ts b/packages/queue/tests/queue.test.ts index 0e10b03c..78d12198 100644 --- a/packages/queue/tests/queue.test.ts +++ b/packages/queue/tests/queue.test.ts @@ -187,6 +187,50 @@ describe('Queue', () => { expect(await queue.size()).toBe(0) }) + it('reports what it did with each job', async () => { + makeResolver() + const queue = new MemoryQueue() + await queue.push(new RecordingJob('watched') as never) + + const seen: string[] = [] + const worker = new Worker(queue).on({ + onProcessing: (job) => { + seen.push(`processing:${job.name()}`) +}, + onProcessed: (job) => { + seen.push(`processed:${job.name()}`) +}, + }) + + await worker.runNextJob() + + expect(seen).toEqual(['processing:RecordingJob', 'processed:RecordingJob']) + }) + + it('reports a failure as retrying, then as permanent', async () => { + makeResolver() + const queue = new MemoryQueue() + await queue.push(new FlakyJob(99) as never) + + const failures: { attempt: number, released: boolean, message: string }[] = [] + const worker = new Worker(queue).on({ + onFailed: (job, error, released) => { + failures.push({ + attempt: job.attempts(), + released, + message: (error as Error).message, + }) + }, + }) + + await worker.daemon({ stopWhenEmpty: true, sleep: 0 }) + + // tries = 3: two releases for another attempt, then a permanent failure. + expect(failures.map((f) => f.released)).toEqual([true, true, false]) + expect(failures.map((f) => f.attempt)).toEqual([1, 2, 3]) + expect(failures[0].message).toMatch(/boom/) + }) + it('stops after maxJobs', async () => { makeResolver() const queue = new MemoryQueue()