diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..89c7120 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,36 @@ +# Changelog + +Notable changes to `@imqueue/job`. Entries start with the first release whose +behavior changes needed a written record; earlier history is in the git log. + +This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). + +## [Unreleased] + +### Added + +- **A job that is lost or not retried is now visible in the log.** Every line + is written through the configured logger on every occurrence, names the + queue and the message id where one exists, and never the job body or an + error text. No control flow, return value or timer was altered. + + - `[JobQueue] push error:` now also covers a write to redis rejected after + `push()` returned, reports at `error` level and carries the queue, the + requested delay and ttl and a failure code. A failure the redis client + delivers twice — through both its command callback and its returned + promise — writes one line. The marker text is unchanged. + - The handler-failure line now states what happens next: `retry in ` or + `no retry`, with the message id, on every failure. + - A retry suppressed because the job's ttl expired, with the message id. + - A re-schedule whose write to redis failed — the promised retry is not + coming, with the message id and a failure code. + + A failure code is never taken from the error as it is: only an allow-listed + code is printed — an `IMQ_`-prefixed framework code, a system `E…` code, a + small integer, a known redis reply code (`WRONGTYPE`, `NOSCRIPT`, + `LOADING`, …) or one of a few known redis-client failure messages mapped to + codes of our own. Everything else, including the error's message, stack and + class name, is reported as `unknown`. A throwing logger can not influence + the queue: every line is written through a contained writer. One deliberate + difference: a logger which throws while an early-rejected push is reported + no longer surfaces that throw, and no longer cancels a re-schedule. diff --git a/index.ts b/index.ts index b42740b..5a00ae4 100644 --- a/index.ts +++ b/index.ts @@ -80,6 +80,138 @@ import IMQ, { type IMQOptions, } from '@imqueue/core'; +/** + * Redis error replies this helper is allowed to quote. The leading token of a + * Redis error reply is a protocol constant, never data — but only these are + * recognised, so an arbitrary upper-case first word of some other error's + * message can never reach the log. + */ +const REDIS_REPLY_CODES: Set = new Set([ + 'ASK', + 'BUSY', + 'BUSYGROUP', + 'CLUSTERDOWN', + 'CROSSSLOT', + 'ERR', + 'EXECABORT', + 'LOADING', + 'MASTERDOWN', + 'MISCONF', + 'MOVED', + 'NOAUTH', + 'NOGROUP', + 'NOPERM', + 'NOPROTO', + 'NOREPLICAS', + 'NOSCRIPT', + 'NOTBUSY', + 'OOM', + 'READONLY', + 'TRYAGAIN', + 'UNBLOCKED', + 'UNKILLABLE', + 'WRONGPASS', + 'WRONGTYPE', +]); + +/** + * Transport failures the redis client reports by message only, mapped to a + * code of our own. The patterns are fixed library strings, so nothing from an + * application error can match them. + */ +const CLIENT_MESSAGE_CODES: Array<[RegExp, string]> = [ + [/^Connection is closed/i, 'CONNECTION_CLOSED'], + [/^Stream connection ended/i, 'STREAM_ENDED'], + [/^Reached the max retries per request limit/i, 'MAX_RETRIES'], + [/^Command timed out/i, 'COMMAND_TIMEOUT'], +]; + +/** + * Shape of an `err.code` this helper accepts: an `IMQ_`-prefixed code of the + * framework itself, or a system errno such as `ECONNREFUSED`. + */ +const SAFE_CODE = /^(IMQ_[A-Z0-9_]{1,48}|E[A-Z]{2,15})$/; + +/** Upper bound of a numeric `err.code`, so that no long number can pass */ +const MAX_NUMERIC_CODE = 65535; + +/** + * Writes one line through the given logger, containing a throwing logger so + * that logging can never influence queue behaviour. Every added line of this + * package goes through here and is written on every occurrence. + * + * @param logger - logger to write the line with + * @param level - logger method to use + * @param message - the line, which must never carry the job body + */ +function logSafe( + logger: ILogger, + level: 'info' | 'warn' | 'error', + message: string, +): void { + try { + logger[level](message); + } catch { + // a failing logger must never influence queue behaviour + } +} + +/** + * Extracts a loggable failure code from an unknown thrown value. + * + * @param err - the caught value, of any shape + * @returns the code, or `unknown` when none can be told safely + * + * @remarks + * Deliberately conservative: only an allow-listed code can come out of here, + * because an application error reaches this helper too and anything of its own + * may carry personal data. Recognised are a framework or system `code`, a + * small numeric `code`, the leading token of a known Redis error reply and a + * known redis-client failure message. Everything else — including the error's + * message, its stack and its class name — yields `unknown`. Never throws. + */ +function errorCode(err: unknown): string { + try { + const code = (err as { code?: unknown } | undefined)?.code; + + if ( + typeof code === 'number' && + Number.isInteger(code) && + code >= 0 && + code <= MAX_NUMERIC_CODE + ) { + return String(code); + } + + if ( + typeof code === 'string' && + (SAFE_CODE.test(code) || REDIS_REPLY_CODES.has(code)) + ) { + return code; + } + + const message = (err as { message?: unknown } | undefined)?.message; + + if (typeof message === 'string') { + const reply = message.split(' ', 1)[0]; + + if (REDIS_REPLY_CODES.has(reply)) { + return reply; + } + + for (const [pattern, mapped] of CLIENT_MESSAGE_CODES) { + if (pattern.test(message)) { + return mapped; + } + } + } + + return 'unknown'; + } catch { + return 'unknown'; + } +} + /** * Everything a job queue needs to connect and behave, given to every constructor * in this package. @@ -549,8 +681,13 @@ export class JobQueuePublisher * @remarks * Fire-and-forget, and worth being deliberate about: this returns * synchronously without waiting for the broker to accept the job, and a failed - * enqueue is reported only by logging `[JobQueue] push error:` through - * {@link JobQueueOptions.logger}. It neither throws nor hands back anything to + * enqueue is reported only by logging `[JobQueue] push error:` at `error` + * level through {@link JobQueueOptions.logger} — with the queue name, the + * requested delay and ttl and the failure code, but never the job body. Both + * causes are covered: the enqueue not starting at all, and the write to + * redis being rejected afterwards, and one failed push writes one such + * line (the underlying transport may add its own write-failure + * diagnostic). It neither throws nor hands back anything to * await, so a caller that must know the job was really enqueued cannot learn it * from here — watch the log, or check for the job's effects. * @@ -560,6 +697,36 @@ export class JobQueuePublisher public push(job: T, options?: PushOptions): JobQueuePublisher { options = options || ({} as PushOptions); + const { delay, ttl } = options; + // both causes of a lost job go through one formatter, so a single + // failure never writes two lines: the send itself rejecting (the + // queue could not be started) and the write to redis being rejected + // later, which core may deliver through both the command callback + // and the returned promise. It must never throw - core calls it + // without a guard of its own - and it never logs the job body + let reported = false; + const report = (err: unknown): void => { + try { + if (reported) { + return; + } + + reported = true; + + const code = errorCode(err); + + logSafe( + this.logger, + 'error', + `[JobQueue] push error: queue ${this.name}, delay ${ + delay === undefined ? 'none' : delay + }, ttl ${ttl === undefined ? 'none' : ttl}, code ${code}`, + ); + } catch { + // logging must never influence the queue + } + }; + this.imq .send( this.name, @@ -571,8 +738,9 @@ export class JobQueuePublisher ...(options.delay ? { delay: options.delay } : {}), }, options.delay, + report, ) - .catch(err => this.logger.log('[JobQueue] push error:', err)); + .catch(report); return this; } @@ -643,7 +811,7 @@ export class JobQueueWorker public onPop(handler: JobQueuePopHandler): JobQueueWorker { this.handler = handler; this.imq.removeAllListeners('message'); - this.imq.on('message', async (message: any) => { + this.imq.on('message', async (message: any, id?: string) => { if (typeof message !== 'object' || !message) { this.logger.warn( '[JobQueue] Invalid message received, skipping:', @@ -655,6 +823,8 @@ export class JobQueueWorker const { job, expire, delay } = message; let rescheduleDelay: number | void | undefined | Promise; + let handlerError: unknown; + let failed = false; try { rescheduleDelay = this.handler?.(job); @@ -670,15 +840,95 @@ export class JobQueueWorker } } catch (err) { rescheduleDelay = delay; - this.logger.log('[JobQueue] Error handling job:', err); + handlerError = err; + failed = true; } - if (typeof expire === 'number' && expire <= Date.now()) { + const expired = typeof expire === 'number' && expire <= Date.now(); + const retrying = + typeof rescheduleDelay === 'number' && rescheduleDelay >= 0; + + // the line is written once the outcome is known, so that it says + // what happens next instead of leaving the reader guessing; the + // handler's error object is never logged - it may quote the job + if (failed) { + const code = errorCode(handlerError); + + // this line existed on every failure before, and each + // occurrence carries its own message id and its own retry + // decision + logSafe( + this.logger, + 'error', + `[JobQueue] Error handling job: queue ${this.name}, ` + + `message ${id || 'unknown'}, code ${code}, ${ + expired || !retrying + ? 'no retry' + : `retry in ${rescheduleDelay} ms` + }`, + ); + } + + if (expired) { + // only when a retry was actually asked for: on a plain + // successful handler this branch is the normal end of a job + if (retrying) { + // one line per expired job: each carries its own message + // id, and the flow is bounded by the expired backlog + logSafe( + this.logger, + 'info', + `[JobQueue] retry suppressed, ttl expired: queue ${ + this.name + }, message ${id || 'unknown'}`, + ); + } + return; // remove job from queue } - if (typeof rescheduleDelay === 'number' && rescheduleDelay >= 0) { - await this.imq.send(this.name, message, rescheduleDelay); + if (retrying) { + // both causes of a failed re-schedule go through one line: + // the send rejecting, and the write to redis being rejected + // afterwards, which core may deliver through both the command + // callback and the returned promise - hence the once-guard. + // The value keeps escaping the handler exactly as it does + // today - it is logged, not swallowed + let reported = false; + const report = (err: unknown): void => { + try { + if (reported) { + return; + } + + reported = true; + + const code = errorCode(err); + + logSafe( + this.logger, + 'error', + `[JobQueue] Job re-schedule failed: queue ${ + this.name + }, message ${id || 'unknown'}, code ${code}`, + ); + } catch { + // logging must never influence the queue + } + }; + + try { + await this.imq.send( + this.name, + message, + rescheduleDelay as number, + report, + ); + } catch (err) { + report(err); + + throw err; + } } }); diff --git a/test/JobQueue.spec.ts b/test/JobQueue.spec.ts index 2f08533..5fc3d63 100644 --- a/test/JobQueue.spec.ts +++ b/test/JobQueue.spec.ts @@ -28,6 +28,22 @@ import './mocks/index.js'; import { logger } from './mocks/index.js'; import JobQueue from '../index.js'; +/** A logger of its own per test, so no other queue's line can be counted */ +const capturing = (): any => { + const captured: any = { info: [], warn: [], error: [] }; + const join = (args: any[]): string => + args.map(arg => String(arg)).join(' '); + + captured.logger = { + log: () => undefined, + info: (...args: any[]) => captured.info.push(join(args)), + warn: (...args: any[]) => captured.warn.push(join(args)), + error: (...args: any[]) => captured.error.push(join(args)), + }; + + return captured; +}; + describe('JobQueue', () => { it('should be a class', () => { assert.equal(typeof JobQueue, 'function'); @@ -217,6 +233,113 @@ describe('JobQueue', () => { spy.restore(); }); + + it('should report a rejected enqueue with its scheduling', async () => { + const cap = capturing(); + const own = new JobQueue({ name: 'Own', logger: cap.logger }); + const send = makeSpy((own as any).imq, 'send').rejects( + new Error('WRONGTYPE customer 000-00-0000'), + ); + + (own as any).handler = () => {}; + await own.push({ ssn: '000-00-0000' }, { ttl: 100, delay: 10 }); + await new Promise(resolve => setImmediate(resolve)); + + assert.equal(cap.error.length, 1); + assert.match(cap.error[0], /\[JobQueue\] push error/); + assert.match(cap.error[0], /queue Own/); + assert.match(cap.error[0], /delay 10/); + assert.match(cap.error[0], /ttl 100/); + assert.match(cap.error[0], /code WRONGTYPE/); + assert.equal(cap.error[0].includes('000-00-0000'), false); + + send.restore(); + await own.destroy(); + }); + + it('should write one line when core delivers one failure twice', async () => { + const cap = capturing(); + const own = new JobQueue({ name: 'Own', logger: cap.logger }); + const send = makeSpy((own as any).imq, 'send'); + + (own as any).handler = () => {}; + await own.push('x'); + + const report = send.args[0][3]; + + assert.equal( + typeof report, + 'function', + 'core must be given an error handler for the write', + ); + assert.doesNotThrow(() => report(new Error('OOM nope'))); + assert.doesNotThrow(() => report(new Error('OOM nope'))); + + const matched = cap.error.filter((one: string) => + /\[JobQueue\] push error/.test(one), + ); + + assert.equal( + matched.length, + 1, + 'a doubly-delivered failure must write one line', + ); + assert.match(matched[0], /delay none/); + assert.match(matched[0], /ttl none/); + assert.match(matched[0], /code OOM/); + + send.restore(); + await own.destroy(); + }); + + it('should report every failed push separately', async () => { + const cap = capturing(); + const own = new JobQueue({ name: 'Own', logger: cap.logger }); + const send = makeSpy((own as any).imq, 'send'); + + (own as any).handler = () => {}; + await own.push('x'); + await own.push('y', { delay: 500 }); + + (send.args[0][3] as any)(new Error('OOM nope')); + (send.args[1][3] as any)(new Error('OOM nope')); + + const matched = cap.error.filter((one: string) => + /\[JobQueue\] push error/.test(one), + ); + + // one line per failed push: no aggregation across pushes + assert.equal(matched.length, 2); + assert.match(matched[0], /delay none/); + assert.match(matched[1], /delay 500/); + + send.restore(); + await own.destroy(); + }); + + it('should survive a broken logger while reporting', async () => { + const broken: any = { + log: () => {}, + info: () => {}, + warn: () => {}, + error: () => { + throw new Error('logger is broken'); + }, + }; + const brokenQueue = new JobQueue({ + name: 'Broken', + logger: broken, + }); + const send = makeSpy((brokenQueue as any).imq, 'send'); + + (brokenQueue as any).handler = () => {}; + await brokenQueue.push('x'); + + assert.doesNotThrow(() => send.args[0][3](new Error('boom'))); + + send.restore(); + await brokenQueue.destroy(); + }); }); describe('onPop', () => { @@ -246,6 +369,8 @@ describe('JobQueue', () => { const deliver = (message: any): Promise => (queue as any).imq.listeners('message')[0](message); + const deliver2 = (message: any, id: string): Promise => + (queue as any).imq.listeners('message')[0](message, id); beforeEach(() => (queue = new JobQueue({ name: 'Test', logger }))); afterEach(async () => { @@ -299,7 +424,12 @@ describe('JobQueue', () => { await deliver(message); assert.equal(send.calledOnce, true); - assert.deepEqual(send.args[0], ['Test', message, 1000]); + assert.deepEqual(send.args[0].slice(0, 3), ['Test', message, 1000]); + assert.equal( + typeof send.args[0][3], + 'function', + 'a late write failure must be reported through core', + ); }); it('should re-schedule with the original delay when the handler throws', async () => { @@ -312,7 +442,243 @@ describe('JobQueue', () => { await deliver(message); assert.equal(send.calledOnce, true); - assert.deepEqual(send.args[0], ['Test', message, 250]); + assert.deepEqual(send.args[0].slice(0, 3), ['Test', message, 250]); + assert.equal(typeof send.args[0][3], 'function'); + }); + + it('should report a failed handler with what happens next', async () => { + const cap = capturing(); + const own = new JobQueue({ name: 'Own', logger: cap.logger }); + const to = (message: any, id: string): Promise => + (own as any).imq.listeners('message')[0](message, id); + + own.onPop(() => { + throw new Error('WRONGTYPE customer 000-00-0000'); + }); + await to({ job: { ssn: '000-00-0000' }, delay: 250 }, 'msg-42'); + + assert.equal(cap.error.length, 1); + assert.match(cap.error[0], /Error handling job/); + assert.match(cap.error[0], /queue Own/); + assert.match(cap.error[0], /message msg-42/); + assert.match(cap.error[0], /code WRONGTYPE/); + assert.match(cap.error[0], /retry in 250 ms/); + assert.equal(cap.error[0].includes('000-00-0000'), false); + + await own.destroy(); + }); + + it('should report every failure of a row, ids and decisions apart', async () => { + const cap = capturing(); + const own = new JobQueue({ name: 'Own', logger: cap.logger }); + const to = (message: any, id: string): Promise => + (own as any).imq.listeners('message')[0](message, id); + + own.onPop(() => { + throw new Error('WRONGTYPE nope'); + }); + // same failure code twice within one minute, different messages + // and different retry decisions: both lines must be written - + // on master every handler failure wrote a line, and keeping + // that is what tells the two messages apart + await to({ job: 'a' }, 'msg-50'); + await to({ job: 'b', delay: 100 }, 'msg-51'); + + assert.equal(cap.error.length, 2); + assert.match(cap.error[0], /message msg-50/); + assert.match(cap.error[0], /no retry/); + assert.match(cap.error[1], /message msg-51/); + assert.match(cap.error[1], /retry in 100 ms/); + + await own.destroy(); + }); + + it('should say a failed handler gets no retry when none is due', async () => { + const cap = capturing(); + const own = new JobQueue({ name: 'Own', logger: cap.logger }); + + own.onPop(() => { + throw new Error('Job error'); + }); + await (own as any).imq.listeners('message')[0]( + { job: 'x' }, + 'msg-43', + ); + + assert.equal(cap.error.length, 1); + assert.match(cap.error[0], /no retry/); + + await own.destroy(); + }); + + it('should report every retry suppressed by an expired ttl', async () => { + const cap = capturing(); + const own = new JobQueue({ name: 'Own', logger: cap.logger }); + + own.onPop(() => 1000); + await (own as any).imq.listeners('message')[0]( + { job: 'x', expire: Date.now() - 1 }, + 'msg-44', + ); + await (own as any).imq.listeners('message')[0]( + { job: 'y', expire: Date.now() - 1 }, + 'msg-45', + ); + + // one line per expired job, each with its own message id + assert.equal(cap.info.length, 2); + assert.match(cap.info[0], /retry suppressed, ttl expired/); + assert.match(cap.info[0], /queue Own/); + assert.match(cap.info[0], /message msg-44/); + assert.match(cap.info[1], /message msg-45/); + + await own.destroy(); + }); + + it('should stay quiet when an expired job asked for no retry', async () => { + const cap = capturing(); + const own = new JobQueue({ name: 'Own', logger: cap.logger }); + + own.onPop(() => undefined); + await (own as any).imq.listeners('message')[0]( + { job: 'x', expire: Date.now() - 1 }, + 'msg-45', + ); + + assert.equal(cap.info.length, 0); + assert.equal(cap.error.length, 0); + + await own.destroy(); + }); + + it('should report a failed re-schedule and still let it escape', async () => { + const error = sandbox.spy(logger, 'error'); + const failure = new Error('WRONGTYPE nope'); + + sandbox.spy((queue as any).imq, 'send').rejects(failure); + queue.onPop(() => 1000); + + await assert.rejects( + () => deliver2({ job: 'x' }, 'msg-46'), + (err: any) => err === failure, + ); + + const line = String(error.args[0]?.[0]); + + assert.match(line, /Job re-schedule failed/); + assert.match(line, /message msg-46/); + assert.match(line, /code WRONGTYPE/); + }); + + it('should write one line when one re-schedule failure comes twice', async () => { + const error = sandbox.spy(logger, 'error'); + const send = sandbox.spy((queue as any).imq, 'send'); + + queue.onPop(() => 1000); + await deliver2({ job: 'x' }, 'msg-47'); + + const report = send.args[0][3]; + + assert.equal(typeof report, 'function'); + assert.doesNotThrow(() => report(new Error('OOM nope'))); + assert.doesNotThrow(() => report(new Error('OOM nope'))); + + const lines = error.args.map((one: any[]) => String(one[0])); + const matched = lines.filter((one: string) => + /Job re-schedule failed/.test(one), + ); + + assert.equal( + matched.length, + 1, + 'a doubly-delivered failure must write one line', + ); + assert.match(matched[0], /code OOM/); + }); + + it('should report every failed re-schedule separately', async () => { + const error = sandbox.spy(logger, 'error'); + const send = sandbox.spy((queue as any).imq, 'send'); + + queue.onPop(() => 1000); + await deliver2({ job: 'x' }, 'msg-48'); + await deliver2({ job: 'y' }, 'msg-49'); + + (send.args[0][3] as any)(new Error('OOM nope')); + (send.args[1][3] as any)(new Error('OOM nope')); + + const lines = error.args.map((one: any[]) => String(one[0])); + const matched = lines.filter((one: string) => + /Job re-schedule failed/.test(one), + ); + + // one line per failed re-schedule, each with its own message id + assert.equal(matched.length, 2); + assert.match(matched[0], /message msg-48/); + assert.match(matched[1], /message msg-49/); + }); + + it('should keep every re-scheduling decision it made before', async () => { + const send = sandbox.spy((queue as any).imq, 'send'); + + // a zero delay is a re-schedule, not a falsy skip + queue.onPop(() => 0); + await deliver2({ job: 'a' }, 'm1'); + + // a resolved promise is awaited and used + queue.onPop(() => Promise.resolve(700)); + await deliver2({ job: 'b' }, 'm2'); + + // a rejected promise falls back to the message's own delay + queue.onPop(() => Promise.reject(new Error('boom'))); + await deliver2({ job: 'c', delay: 250 }, 'm3'); + + // anything not a number is not a re-schedule + queue.onPop(() => 'soon' as any); + await deliver2({ job: 'd' }, 'm4'); + + // a non-numeric ttl never suppresses anything + queue.onPop(() => 900); + await deliver2({ job: 'e', expire: 'yesterday' }, 'm5'); + + assert.deepEqual( + send.args.map((one: any[]) => one[2]), + [0, 700, 250, 900], + ); + }); + + it('should not let a broken logger cancel a re-schedule', async () => { + // named difference from the previous release: the line is written + // through a contained writer, so a logger which throws no longer + // takes the re-scheduling down with it + const broken: any = { + log: () => {}, + info: () => {}, + warn: () => {}, + error: () => { + throw new Error('logger is broken'); + }, + }; + const brokenQueue = new JobQueue({ + name: 'Broken', + logger: broken, + }); + const send = makeSpy((brokenQueue as any).imq, 'send'); + + brokenQueue.onPop(() => { + throw new Error('Job error'); + }); + + await (brokenQueue as any).imq.listeners('message')[0]( + { job: 'x', delay: 100 }, + 'm6', + ); + + assert.equal(send.calledOnce, true); + assert.equal(send.args[0][2], 100); + + send.restore(); + await brokenQueue.destroy(); }); it('should not re-schedule a job whose ttl has passed', async () => {