diff --git a/CHANGELOG.md b/CHANGELOG.md index b3d1b60..174a396 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,60 @@ 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 + +- **Silent failures of the transport are now reported without `verbose`.** Every + line below is written through the configured logger, carries the queue, + channel or host it is about and never the message payload, the arguments, a + redis key or an error text. Nothing is scheduled and no timer is added. + + - A write to redis rejected inside `send()` — the caller already holds the + message id and gets no rejection, so this was observable only through the + optional `errorHandler`, which most callers do not pass. Reported as a + failure episode per queue instance: the first rejected write is logged + with its operation, message id and code, further rejections are only + counted, and the first successful write logs the recovery together with + that count. A failure a redis client delivers twice — through both its + command callback and its returned promise — is counted once, while + `errorHandler` keeps being invoked per delivery, exactly as before. + - Safe reading of a queue ending on anything other than a planned stop, + reconnect or destroy. A planned stop stays quiet, as before. + - A failure of the periodic watcher-existence check itself; the failures of + delayed-message processing and of watcher initialization were already + logged and are not duplicated. + - A subscription being established and being restored after a reconnect, and + a failed reconnection attempt — the absence of the restore line after a + reconnect is what makes a lost subscription provable. + - Safe-delivery maintenance disabling itself for good when the writer + connection is gone. + - Messages of expired worker leases being re-queued — aggregated per + processing pass into one line per destination queue with a count — and a + worker key that could not be deleted: the two causes of a duplicate + delivery. + - Keys removed by the built-in cleanup, with the number of candidates and the + number actually deleted. + - A publish whose channel has no subscribers, on entering that state, and, in + a clustered queue, a publish with no server to publish to at all. + - In a clustered queue: round-robin having no available instance left, and a + joining server failing to start or to subscribe, with the host and the + phase. + + No control flow, return value, redis round-trip or timer was altered, and no + new public API was added. One deliberate difference: the line reporting a + worker key that could not be deleted is now written through a contained + writer, so a logger which itself throws can no longer surface that throw as + an unhandled rejection — every line of this change is required to be unable + to influence queue behaviour. + + 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`. + ## [3.3.3] - 2026-08-18 ### Fixed diff --git a/src/ClusteredRedisQueue.ts b/src/ClusteredRedisQueue.ts index c25e51b..e0c3158 100644 --- a/src/ClusteredRedisQueue.ts +++ b/src/ClusteredRedisQueue.ts @@ -23,7 +23,7 @@ */ import { EventEmitter } from 'node:events'; import { type InitializedCluster } from './ClusterManager.js'; -import { buildOptions, copyEventEmitter } from './helpers/index.js'; +import { buildOptions, copyEventEmitter, errorCode } from './helpers/index.js'; import { DEFAULT_IMQ_OPTIONS, type EventMap, @@ -137,6 +137,18 @@ export class ClusteredRedisQueue */ private imqLength: number = 0; + /** + * True while round-robin has no available instance left to pick, so that + * the condition is reported on entry only and not on every send + */ + private noneAvailable: boolean = false; + + /** + * True while a publish had no server to publish to, so that the condition + * is reported on entry only and not on every publish + */ + private noPublishTargets: boolean = false; + /** * Template EventEmitter instance used to replicate queue EventEmitters when * dynamically modifying the cluster @@ -332,11 +344,25 @@ export class ClusteredRedisQueue if (candidate.available) { this.currentQueue = index + 1; + this.noneAvailable = false; return candidate; } } + // every instance reports its connection as not ready, so the message + // goes to a host known to be down. Reported on entering the state + // only, and the flag is cleared by the first successful pick above, + // so a second outage after a recovery is visible again + if (!this.noneAvailable) { + this.noneAvailable = true; + this.logLine( + 'warn', + `no available instance out of ${count}, sending to an ` + + 'instance which is known to be down', + ); + } + this.currentQueue = start + 1; return this.imqs[start]; @@ -455,6 +481,29 @@ export class ClusteredRedisQueue } } + /** + * Writes an unconditional line through this cluster's logger, in the + * format `verbose()` uses. + * + * @param level - logger method to write the line with + * @param message - the line, which must never carry message payload, + * call arguments, raw redis keys or an error text + * + * @remarks + * Never throws: a broken logger must not be able to change what the + * cluster does. Every call site of this reports a state transition or a + * lifecycle event, so no rate limiting is needed. + */ + private logLine(level: 'info' | 'warn' | 'error', message: string): void { + try { + this.logger[level]( + `[IMQ-CORE][ClusteredRedisQueue][${this.name}]: ${message}`, + ); + } catch { + // a failing logger must never influence cluster behaviour + } + } + /** * Batch imq action processing on all registered imqs at once * @@ -751,8 +800,9 @@ export class ClusteredRedisQueue * * Publication is not atomic — if any host has no writer connection the call * rejects even though other hosts may already have published. On an empty - * cluster it resolves without publishing anything, and unlike `send()` it does - * not wait for a server to appear. + * cluster it resolves without publishing anything — reporting that through + * the logger on entering the state — and unlike `send()` it does not wait + * for a server to appear. */ public async publish(data: JsonObject, toName?: string): Promise { const promises: Array> = []; @@ -761,6 +811,23 @@ export class ClusteredRedisQueue promises.push(imq.publish(data, toName)); } + // an empty cluster resolves an empty Promise.all, so the caller is + // told the event went out while nothing was published at all. + // Reported on entering the state only + if (!promises.length) { + if (!this.noPublishTargets) { + this.noPublishTargets = true; + this.logLine( + 'error', + `nothing published to channel ${ + toName || this.name + }: knownServers=0`, + ); + } + } else { + this.noPublishTargets = false; + } + await Promise.all(promises); } @@ -938,15 +1005,43 @@ export class ClusteredRedisQueue `Initializing queue with state: ${JSON.stringify(this.state)}`, ); + // both failures are reported here, inside the function, and the + // value is re-thrown as it was: the caller starts this without + // awaiting it, so a new .catch() would either swallow the failure or + // add a second unhandled rejection if (this.state.started) { - await imq.start(); + try { + await imq.start(); + } catch (err) { + this.logLine( + 'error', + `server ${imq.redisKey} failed to start, code ${errorCode( + err, + )}: the node is not ready to serve queues`, + ); + + throw err; + } } if (this.state.subscription) { - await imq.subscribe( - this.state.subscription.channel, - this.state.subscription.handler, - ); + try { + await imq.subscribe( + this.state.subscription.channel, + this.state.subscription.handler, + ); + } catch (err) { + this.logLine( + 'error', + `server ${imq.redisKey} failed to subscribe to channel ${ + this.state.subscription.channel + }, code ${errorCode( + err, + )}: events from this node will never arrive`, + ); + + throw err; + } } } diff --git a/src/RedisQueue.ts b/src/RedisQueue.ts index 2c78e0c..4aab781 100644 --- a/src/RedisQueue.ts +++ b/src/RedisQueue.ts @@ -43,6 +43,8 @@ import { sha1, unpack, envInt, + errorCode, + LOG_MAX_KEYS, } from './helpers/index.js'; import Redis from './redis.js'; @@ -283,6 +285,23 @@ export class RedisQueue */ private watcherCheckBusy: boolean = false; + /** + * Number of rejected writes in the current failure episode of this + * instance's writer, zero while writes succeed. The first rejection of an + * episode is logged, the rest only counted, and the first successful + * write logs the recovery together with this count. + */ + private rejectedWrites: number = 0; + + /** + * Channels a publish currently finds no subscribers on, so that the + * condition is reported on entry only and not on every publish. Bounded + * by {@link LOG_MAX_KEYS}: channel names may arrive as unique values, so + * names above the bound are not remembered and their publishes are + * reported every time. + */ + private readonly noSubscribers: Set = new Set(); + /** * Connected client keys seen during the previous cleanup sweep. Used * to give temporarily disconnected clients one sweep of grace before @@ -419,6 +438,63 @@ export class RedisQueue } } + /** + * Writes an unconditional line through this queue's logger, in the format + * `verbose()` uses. + * + * @param level - logger method to write the line with + * @param message - the line, which must never carry message payload, + * call arguments, raw redis keys or an error text + * + * @remarks + * Never throws: a broken logger must not be able to change what the queue + * does. + */ + private logLine(level: 'info' | 'warn' | 'error', message: string): void { + try { + this.logger[level](`[IMQ-CORE][${this.name}]: ${message}`); + } catch { + // a failing logger must never influence queue behaviour + } + } + + /** + * Records one logical rejected write of this instance's writer: the first + * rejection of a failure episode writes the given line, every following + * one only increments the episode counter, and the counter is reported by + * {@link RedisQueue.recordWriteSuccess} when a write succeeds again. + * + * @param message - the line, under the same constraints as + * {@link RedisQueue.logLine} + */ + private recordWriteFailure(message: string): void { + this.rejectedWrites++; + + if (this.rejectedWrites === 1) { + this.logLine('error', message); + } + } + + /** + * Closes the current write-failure episode, if one is open: reports how + * many writes were rejected in it and resets the counter. The counter is + * reset before the logger is touched, so a broken logger cannot keep the + * episode open forever. + */ + private recordWriteSuccess(): void { + if (this.rejectedWrites === 0) { + return; + } + + const rejected = this.rejectedWrites; + + this.rejectedWrites = 0; + this.logLine( + 'info', + `outbound writes resumed after ${rejected} rejected writes`, + ); + } + /** * Creates a subscription channel over redis and sets up channel * data read handler. The effective Redis channel is `:`. @@ -470,6 +546,9 @@ export class RedisQueue this.subscriptionHandlers.push(handler); this.verbose(`Subscribed to ${channel} channel`); + // a lifecycle fact, not an alarm: flow continuation hangs on this + // subscription, so its absence after a reconnect must be provable + this.logLine('info', `subscribed to channel ${channel}`); } /** @@ -523,6 +602,10 @@ export class RedisQueue } this.verbose(`Restored subscription to ${this.subscriptionName}`); + this.logLine( + 'info', + `restored subscription to channel ${this.subscriptionName}`, + ); } /** @@ -585,8 +668,9 @@ export class RedisQueue * allowed in any {@link IMQMode}, including `WORKER`. * * The payload is always plain JSON — {@link IMQOptions.useGzip} applies to - * queue messages only. Redis pub/sub drops the message silently when nobody - * is subscribed. + * queue messages only. Redis pub/sub drops the message when nobody is + * subscribed; on entering that state the queue writes a warning through + * its logger, so the drop is no longer silent. */ public async publish(data: JsonObject, toName?: string): Promise { if (!this.writer) { @@ -596,7 +680,33 @@ export class RedisQueue const jsonData = JSON.stringify(data); const name = toName || this.name; - await this.writer.publish(`${this.options.prefix}:${name}`, jsonData); + const receivers = await this.writer.publish( + `${this.options.prefix}:${name}`, + jsonData, + ); + + // redis replies with the number of subscribers which received the + // event: zero means nobody did, and events are how a consumer learns + // to continue its work. Reported on entering the state only, and only + // on a strict zero - a client whose reply is not a number is left + // alone rather than coerced, so reading it can neither throw nor + // invent a warning + if (receivers === 0) { + if (!this.noSubscribers.has(name)) { + if (this.noSubscribers.size < LOG_MAX_KEYS) { + this.noSubscribers.add(name); + } + + this.logLine( + 'warn', + `published to channel ${name} on host ${ + this.redisKey + } with no subscribers`, + ); + } + } else { + this.noSubscribers.delete(name); + } this.verbose(`Published message to ${name} channel, data: ${jsonData} `); @@ -745,7 +855,25 @@ export class RedisQueue this.watcherCheckBusy = true; try { - if (!(await this.watcherCount())) { + let watchers: number; + + try { + watchers = await this.watcherCount(); + } catch (err) { + // the only silent failure of this tick: initWatcher() and + // processDelayed() log their own errors unconditionally, so + // wrapping the whole body would just duplicate them. The + // value is re-thrown, leaving control flow as it was + // bounded by watcherCheckDelay: this tick runs on an + // interval, so a persistent failure repeats at that pace + const code = errorCode(err); + + this.logLine('warn', `watcher check failed, code ${code}`); + + throw err; + } + + if (!watchers) { await this.initWatcher(); } @@ -779,9 +907,12 @@ export class RedisQueue * schedule — the message is released by the watcher, so availability * may lag by up to {@link IMQOptions.watcherCheckDelay}. A delay of * `0` or `undefined` sends immediately. - * @param errorHandler - invoked when the write to Redis fails; this is the - * only way to observe such a failure, since the returned promise does - * not reject for it + * @param errorHandler - invoked when the write to Redis fails; the returned + * promise does not reject for it, so this is the only programmatic + * way to observe such a failure. The failure is also reported + * through the queue's logger: the first rejected write of a failure + * episode is logged, the rest are counted, and the first successful + * write logs the recovery with that count * @returns the identifier assigned to the message. It is generated locally * before the write is issued, so it is available even if the write * later fails. @@ -819,10 +950,26 @@ export class RedisQueue const data: IMessage = { id, message, from: this.name }; const key = `${this.options.prefix}:${toQueue}`; const packet = this.pack(data); + const countedOps = new Set(); const onWriteError = (error: unknown, op: string): void => { if (error) { this.verbose(`Writer ${op} error: ${error}`); + // the caller already holds the message id and gets no + // rejection, so without this line a rejected write is + // observable through errorHandler only - and most callers + // pass none. A client may deliver the same failure through + // both its callback and its returned promise: the episode + // counts each logical failure once, while errorHandler keeps + // being invoked per delivery, exactly as it always was + if (!countedOps.has(op)) { + countedOps.add(op); + this.recordWriteFailure( + `write to queue ${toQueue} rejected on ${op}, ` + + `message ${id}, code ${errorCode(error)}`, + ); + } + if (errorHandler) { errorHandler( error instanceof Error @@ -858,6 +1005,11 @@ export class RedisQueue return; } + + // a delayed send is complete only after both + // ZADD and SET succeeded, so the episode is + // closed here and not on the ZADD alone + this.recordWriteSuccess(); }, ) .catch((err: unknown) => onWriteError(err, 'SET')); @@ -870,6 +1022,8 @@ export class RedisQueue (err?: Error | null) => { if (err) { onWriteError(err, 'LPUSH'); + } else { + this.recordWriteSuccess(); } }, ); @@ -1428,6 +1582,14 @@ export class RedisQueue } catch (err) { this.reconnecting[channel] = false; this.verbose(`Reconnect ${channel} failed: ${err}`); + // bounded by the exponential reconnection backoff, so a redis + // which stays down cannot make this line flood the log + this.logLine( + 'warn', + `reconnect of the ${channel} channel failed, code ${errorCode( + err, + )}`, + ); this.scheduleReconnect(channel); } } @@ -1651,29 +1813,70 @@ export class RedisQueue return; } + const requeued = new Map(); + this.verbose( `Watching ${keys.length} keys: ${keys .map(key => `"${key}"`) .join(', ')}`, ); - for (const key of keys) { - const kp: string[] = key.split(':'); + try { + for (const key of keys) { + const kp: string[] = key.split(':'); + + // the last key segment is the worker's lease deadline: only + // re-queue messages of workers whose lease has expired (the + // worker died mid-processing); fresh leases belong to live + // workers and must not be touched + if (Number(kp.pop()) >= now) { + continue; + } - // the last key segment is the worker's lease deadline: only - // re-queue messages of workers whose lease has expired (the - // worker died mid-processing); fresh leases belong to live - // workers and must not be touched - if (Number(kp.pop()) >= now) { - continue; - } + const target = `${kp.shift()}:${kp.shift()}`; + const moved = await this.writer.lmove( + key, + target, + 'RIGHT', + 'LEFT', + ); - await this.writer.lmove( - key, - `${kp.shift()}:${kp.shift()}`, - 'RIGHT', - 'LEFT', - ); + // a non-empty result means a message whose lease expired is being + // delivered a second time - either its worker died or it took + // longer than the lease ttl. This is one of the two explanations + // a handler can be given for a duplicate, the other being a + // worker key that could not be deleted. The packed message is + // never unpacked here: it carries the payload, and the worker key + // is never logged either + if (moved) { + // the raw target is a redis key and is never printed: when + // the exact prefix cannot be stripped off safely (a prefix + // carrying ':' breaks the segment arithmetic above), the + // queue is reported as unknown rather than leaked + const stripped = `${this.options.prefix}:`; + const queue = + target.startsWith(stripped) && + target.length > stripped.length + ? target.slice(stripped.length) + : 'unknown'; + + requeued.set(queue, (requeued.get(queue) || 0) + 1); + } + } + } finally { + // one line per queue for the whole pass: a backlog of expired + // leases is reported as a count instead of a line per message. + // Written in a finally so that re-queues which did happen stay + // reported even when a later move of the pass throws - the + // exception itself keeps escaping exactly as before + for (const [queue, count] of requeued) { + this.logLine( + 'warn', + `re-queued ${count} messages of expired leases to queue ${ + queue + }`, + ); + } } } @@ -1833,6 +2036,18 @@ export class RedisQueue */ private async runSafeCheck(): Promise { if (!this.writer) { + // one line per interval instance, because the interval is + // dropped right below and re-armed only by a new watcher + // connection: from here on nothing recovers abandoned messages + // and nothing prunes orphaned keys + const safe = !!this.options.safeDelivery; + const cleanup = !!this.options.cleanup; + + this.logLine( + 'warn', + 'safe delivery maintenance stopped: no writer connection, ' + + `safeDelivery ${safe}, cleanup ${cleanup}`, + ); this.cleanSafeCheckInterval(); return; @@ -1933,13 +2148,29 @@ export class RedisQueue } if (keysToRemove.length) { - await this.writer.del(...keysToRemove); + const removed = await this.writer.del(...keysToRemove); this.verbose( `Keys ${keysToRemove .map(k => `"${k}"`) .join(', ')} were successfully removed!`, ); + + // deleting the keys of a queue considered abandoned is + // destructive - a client which was disconnected for two + // sweeps loses its queue together with a response it still + // waits for. Neither the keys nor the filter are logged: + // they carry application names + if (typeof removed === 'number' && removed > 0) { + // bounded by the maintenance interval: one line per + // cleanup pass at most, and it already aggregates counts + this.logLine( + 'warn', + `cleanup removed ${removed} of ${ + keysToRemove.length + } candidate keys`, + ); + } } } catch (err) { this.logger.warn('Clean-up error occurred:', err); @@ -2025,8 +2256,40 @@ export class RedisQueue 'LEFT', timeout, ); - } catch { - // reader connection ended (stop/reconnect) + } catch (err) { + // a closed or ended reader connection is the planned case - + // the queue is stopping, reconnecting or being destroyed, + // and stop() drops the reader socket on purpose - so it + // stays quiet. Anything else ends safe reading for good, + // which must not be silent: the process stays alive and + // simply consumes nothing + let planned = this.destroyed || !this.reader; + + if (!planned) { + try { + planned = + err instanceof Error && + /Stream connection ended|Connection is closed/i.test( + err.message, + ); + } catch { + // a message getter which throws is read as an + // unexpected failure - before this catch existed, + // such a value would have escaped readSafe() as an + // unhandled rejection + planned = false; + } + } + + if (!planned) { + this.logLine( + 'warn', + `safe reading of queue ${ + this.name + } stopped, code ${errorCode(err)}`, + ); + } + break; } @@ -2037,9 +2300,18 @@ export class RedisQueue try { this.process([key, msg]); - this.writer - .del(workerKey) - .catch(e => this.logger.warn('OnReadSafe: del error', e)); + this.writer.del(workerKey).catch((err: unknown) => + // an undeleted worker key is re-queued by the watcher + // later, which is the other cause of a duplicate. The + // key itself is never logged: it carries the client + // queue name, the lease uuid and its deadline + this.logLine( + 'warn', + `OnReadSafe: del error, queue ${ + this.name + }, code ${errorCode(err)}`, + ), + ); } catch (err) { // a single message failure must never kill the read loop this.emitError('OnReadSafe', 'safe reader failed', err); diff --git a/src/helpers/index.ts b/src/helpers/index.ts index b9008a8..6e85e8a 100644 --- a/src/helpers/index.ts +++ b/src/helpers/index.ts @@ -29,3 +29,4 @@ export * from './unpack.js'; export * from './escapeRegExp.js'; export * from './copyEventEmitter.js'; export * from './envInt.js'; +export * from './logging.js'; diff --git a/src/helpers/logging.ts b/src/helpers/logging.ts new file mode 100644 index 0000000..0323f2e --- /dev/null +++ b/src/helpers/logging.ts @@ -0,0 +1,141 @@ +/*! + * Helper: logging + * + * I'm Queue Software Project + * Copyright (C) 2025 imqueue.com + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + * + * If you want to use this code in a closed source (commercial) project, you can + * purchase a proprietary commercial license. Please contact us at + * to get commercial licensing options. + */ +/** + * Upper bound on the number of channel names the no-subscriber transition + * state of a queue remembers. Channel names arrive as an unbounded stream of + * unique client-queue names, so the set must not grow with them: names above + * the bound are simply not remembered. + */ +export const LOG_MAX_KEYS = 128; + +/** + * 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; + +/** + * 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. + */ +export 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'; + } +} diff --git a/test/unit/ClusteredRedisQueue.spec.ts b/test/unit/ClusteredRedisQueue.spec.ts index 52359d8..d8c3435 100644 --- a/test/unit/ClusteredRedisQueue.spec.ts +++ b/test/unit/ClusteredRedisQueue.spec.ts @@ -748,3 +748,196 @@ describe('ClusteredRedisQueue fan-out helpers', () => { await cq.destroy(); }); }); + +/** Logger which keeps every line it was given, per level */ +const capturing = (): any => { + const join = (args: any[]): string => + args.map(arg => String(arg)).join(' '); + const captured: any = { info: [], warn: [], error: [] }; + + 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; +}; + +const matching = (lines: string[], rx: RegExp): string[] => + lines.filter(line => rx.test(line)); + +const unavailable = (imq: any, available: boolean): void => { + Object.defineProperty(imq, 'available', { + configurable: true, + get: () => available, + }); +}; + +describe('ClusteredRedisQueue instance selection logging', () => { + afterEach(() => mock.restoreAll()); + + it('warns on entering the state where no instance is available', async () => { + const cap = capturing(); + const cq: any = new ClusteredRedisQueue('SelectNone', { + ...clusterConfig, + logger: cap.logger, + }); + + cq.imqs.forEach((imq: any) => unavailable(imq, false)); + + cq.selectQueue(); + cq.selectQueue(); + + const lines = matching(cap.warn, /no available instance/); + + assert.equal(lines.length, 1); + assert.match(lines[0], /out of 2/); + + unavailable(cq.imqs[0], true); + cq.selectQueue(); + + assert.equal(matching(cap.warn, /no available instance/).length, 1); + + unavailable(cq.imqs[0], false); + cq.selectQueue(); + + assert.equal(matching(cap.warn, /no available instance/).length, 2); + + await cq.destroy(); + }); + + it('stays quiet while an instance is available', async () => { + const cap = capturing(); + const cq: any = new ClusteredRedisQueue('SelectOk', { + ...clusterConfig, + logger: cap.logger, + }); + + cq.selectQueue(); + cq.selectQueue(); + + assert.equal(matching(cap.warn, /no available instance/).length, 0); + + await cq.destroy(); + }); +}); + +describe('ClusteredRedisQueue publish visibility', () => { + afterEach(() => mock.restoreAll()); + + it('reports an event published to an empty cluster once', async () => { + const cap = capturing(); + const clusterManager = new (ClusterManager as any)(); + const cq: any = new ClusteredRedisQueue('PubEmpty', { + clusterManagers: [clusterManager], + logger: cap.logger, + }); + + await cq.publish({ ssn: '000-00-0000' }, 'FlowEvents'); + await cq.publish({ ssn: '000-00-0000' }, 'FlowEvents'); + + const lines = matching(cap.error, /nothing published/); + + assert.equal(lines.length, 1); + assert.match(lines[0], /FlowEvents/); + assert.match(lines[0], /knownServers=0/); + assert.equal(lines[0].includes('000-00-0000'), false); + + await cq.destroy(); + }); + + it('stays quiet when the cluster has servers to publish to', async () => { + const cap = capturing(); + const cq: any = new ClusteredRedisQueue('PubServers', { + ...clusterConfig, + logger: cap.logger, + }); + + mock.method( + RedisQueue.prototype as any, + 'publish', + async () => undefined, + ); + + await cq.publish({ a: 1 }); + + assert.equal(matching(cap.error, /nothing published/).length, 0); + + await cq.destroy(); + }); +}); + +describe('ClusteredRedisQueue joining-server failure logging', () => { + afterEach(() => mock.restoreAll()); + + it('names the host and the phase when a joining server cannot start', async () => { + const cap = capturing(); + const cq: any = new ClusteredRedisQueue('JoinStartFail', { + ...clusterConfig, + logger: cap.logger, + }); + const boom = Object.assign(new Error('host is down'), { + code: 'ECONNREFUSED', + }); + + cq.state.started = true; + + let thrown: any; + + try { + await cq.initializeQueue({ + redisKey: '127.0.0.1:9999', + start: () => Promise.reject(boom), + }); + } catch (err) { + thrown = err; + } + + const lines = matching(cap.error, /failed to start/); + + assert.equal(thrown, boom, 'the same value must be re-thrown'); + assert.equal(lines.length, 1); + assert.match(lines[0], /127\.0\.0\.1:9999/); + assert.match(lines[0], /ECONNREFUSED/); + assert.equal(lines[0].includes('host is down'), false); + + await cq.destroy(); + }); + + it('names the channel when a joining server cannot subscribe', async () => { + const cap = capturing(); + const cq: any = new ClusteredRedisQueue('JoinSubFail', { + ...clusterConfig, + logger: cap.logger, + }); + const boom = new Error('NOPERM no permissions'); + + cq.state.started = false; + cq.state.subscription = { + channel: 'FlowEvents', + handler: () => undefined, + }; + + let thrown: any; + + try { + await cq.initializeQueue({ + redisKey: '127.0.0.1:9999', + subscribe: () => Promise.reject(boom), + }); + } catch (err) { + thrown = err; + } + + const lines = matching(cap.error, /failed to subscribe/); + + assert.equal(thrown, boom, 'the same value must be re-thrown'); + assert.equal(lines.length, 1); + assert.match(lines[0], /FlowEvents/); + assert.match(lines[0], /NOPERM/); + + await cq.destroy(); + }); +}); diff --git a/test/unit/RedisQueue.spec.ts b/test/unit/RedisQueue.spec.ts index 8eac876..0ccff27 100644 --- a/test/unit/RedisQueue.spec.ts +++ b/test/unit/RedisQueue.spec.ts @@ -38,7 +38,7 @@ import { } from 'node:test'; import { Redis } from 'ioredis'; import { RedisQueue, IMQMode } from '../../src/index.js'; -import { escapeRegExp, sha1 } from '../../src/helpers/index.js'; +import { escapeRegExp, pack, sha1 } from '../../src/helpers/index.js'; import { makeLogger } from '../helpers/index.js'; import { logger, RedisClientMock } from '../mocks/index.js'; @@ -2142,3 +2142,833 @@ describe('RedisQueue remaining guards', () => { await rq.destroy().catch(() => undefined); }); }); + +/** Logger which keeps every line it was given, per level */ +interface Captured { + logger: any; + info: string[]; + warn: string[]; + error: string[]; +} + +const capturing = (): Captured => { + const join = (args: any[]): string => + args.map(arg => String(arg)).join(' '); + const captured: Captured = { + info: [], + warn: [], + error: [], + logger: undefined, + }; + + 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; +}; + +const matching = (lines: string[], rx: RegExp): string[] => + lines.filter(line => rx.test(line)); + +describe('RedisQueue write failure logging', () => { + afterEach(() => mock.restoreAll()); + + it('logs the first rejected write of an episode, without payload', async t => { + const cap = capturing(); + const rq: any = new RedisQueue(uuid(), { logger: cap.logger }); + + await rq.start(); + t.after(() => rq.destroy(true).catch(() => undefined)); + mock.method(rq.writer, 'lpush', (_k: any, _v: any, cb: any) => { + cb(new Error('WRONGTYPE against a key holding secret-payload')); + + return 0; + }); + + const handled: Error[] = []; + const id = await rq.send( + 'WriteTarget', + { pan: '4111111111111111' }, + undefined, + (err: Error) => handled.push(err), + ); + const lines = matching(cap.error, /write to queue/); + + assert.equal(typeof id, 'string', 'the returned id must not change'); + assert.equal(handled.length, 1, 'errorHandler must still be called'); + assert.equal(lines.length, 1); + assert.match(lines[0], /WriteTarget/); + assert.match(lines[0], /LPUSH/); + assert.match(lines[0], /WRONGTYPE/); + assert.match(lines[0], new RegExp(id)); + assert.equal(lines[0].includes('4111111111111111'), false); + assert.equal(lines[0].includes('secret-payload'), false); + }); + + it('keeps later failures silent until a write succeeds again', async t => { + const cap = capturing(); + const rq: any = new RedisQueue(uuid(), { logger: cap.logger }); + + await rq.start(); + t.after(() => rq.destroy(true).catch(() => undefined)); + + let fail = true; + + mock.method(rq.writer, 'lpush', (_k: any, _v: any, cb: any) => { + cb( + fail + ? Object.assign(new Error('nope'), { code: 'EPIPE' }) + : null, + ); + + return 0; + }); + + await rq.send('WriteTarget', { a: 1 }); + await rq.send('WriteTarget', { a: 2 }); + await rq.send('WriteTarget', { a: 3 }); + + assert.equal(matching(cap.error, /write to queue/).length, 1); + assert.equal(matching(cap.info, /writes resumed/).length, 0); + + fail = false; + await rq.send('WriteTarget', { a: 4 }); + + const resumed = matching(cap.info, /writes resumed/); + + assert.equal(resumed.length, 1); + assert.match(resumed[0], /after 3 rejected writes/); + + fail = true; + await rq.send('WriteTarget', { a: 5 }); + + assert.equal( + matching(cap.error, /write to queue/).length, + 2, + 'a failure after the recovery opens a new episode', + ); + }); + + it('counts one rejection once when callback and promise both fire', async t => { + const cap = capturing(); + const rq: any = new RedisQueue(uuid(), { logger: cap.logger }); + + await rq.start(); + t.after(() => rq.destroy(true).catch(() => undefined)); + + const failure = Object.assign(new Error('nope'), { code: 'EPIPE' }); + + mock.method(rq.writer, 'lpush', (_k: any, _v: any, cb: any) => { + cb(failure); + + return Promise.reject(failure); + }); + + const handled: Error[] = []; + + await rq.send('WriteTarget', { a: 1 }, undefined, (err: Error) => + handled.push(err), + ); + await new Promise(resolve => setImmediate(resolve)); + + assert.equal( + matching(cap.error, /write to queue/).length, + 1, + 'one logical failure must open the episode once', + ); + assert.equal( + handled.length, + 2, + 'errorHandler keeps being invoked per delivery, as it always was', + ); + + mock.method(rq.writer, 'lpush', (_k: any, _v: any, cb: any) => { + cb(null); + + return 0; + }); + + await rq.send('WriteTarget', { a: 2 }); + + const resumed = matching(cap.info, /writes resumed/); + + assert.equal(resumed.length, 1); + assert.match( + resumed[0], + /after 1 rejected writes/, + 'the double delivery must count as one rejected write', + ); + }); + + it('does not resume a delayed-write episode until SET succeeds', async t => { + const cap = capturing(); + const rq: any = new RedisQueue(uuid(), { logger: cap.logger }); + + await rq.start(); + t.after(() => rq.destroy(true).catch(() => undefined)); + + let failSet = true; + + mock.method(rq.writer, 'zadd', (...args: any[]) => { + args[args.length - 1](null); + + return 0; + }); + mock.method(rq.writer, 'set', (...args: any[]) => { + args[args.length - 1]( + failSet + ? Object.assign(new Error('nope'), { code: 'EPIPE' }) + : null, + ); + + return Promise.resolve(); + }); + + await rq.send('DelayTarget', { a: 1 }, 1000); + + assert.equal( + matching(cap.info, /writes resumed/).length, + 0, + 'a successful ZADD alone must not close the episode', + ); + assert.equal(matching(cap.error, /write to queue/).length, 1); + + failSet = false; + await rq.send('DelayTarget', { a: 2 }, 1000); + + assert.equal(matching(cap.info, /writes resumed/).length, 1); + }); + + it('reopens the episode even when the recovery logger throws', async t => { + const errors: string[] = []; + const broken: any = { + log: () => {}, + info: (...args: any[]) => { + if (/writes resumed/.test(args.join(' '))) { + throw new Error('logger is broken'); + } + }, + warn: () => {}, + error: (...args: any[]) => errors.push(args.join(' ')), + }; + const rq: any = new RedisQueue(uuid(), { logger: broken }); + + await rq.start(); + t.after(() => rq.destroy(true).catch(() => undefined)); + + let fail = true; + + mock.method(rq.writer, 'lpush', (_k: any, _v: any, cb: any) => { + cb( + fail + ? Object.assign(new Error('nope'), { code: 'EPIPE' }) + : null, + ); + + return 0; + }); + + await rq.send('WriteTarget', { a: 1 }); + + fail = false; + // the recovery line is swallowed by the throwing logger, but the + // episode counter was reset before the logger was touched + await rq.send('WriteTarget', { a: 2 }); + + fail = true; + await rq.send('WriteTarget', { a: 3 }); + + assert.equal( + errors.filter(one => /write to queue/.test(one)).length, + 2, + 'a broken recovery logger must not keep the episode open', + ); + }); + + it('logs a rejected delayed write with the failing operation', async t => { + const cap = capturing(); + const rq: any = new RedisQueue(uuid(), { logger: cap.logger }); + + await rq.start(); + t.after(() => rq.destroy(true).catch(() => undefined)); + mock.method(rq.writer, 'zadd', (...args: any[]) => { + args[args.length - 1](new Error('OOM command not allowed')); + + return false; + }); + + await rq.send('DelayTarget', { a: 1 }, 1000); + + const lines = matching(cap.error, /write to queue/); + + assert.equal(lines.length, 1); + assert.match(lines[0], /ZADD/); + assert.match(lines[0], /OOM/); + }); +}); + +describe('RedisQueue safe reading interruption logging', () => { + afterEach(() => mock.restoreAll()); + + it('warns when safe reading ends on an unexpected failure', async t => { + const cap = capturing(); + const rq: any = new RedisQueue(uuid(), { + logger: cap.logger, + safeDelivery: true, + safeDeliveryTtl: 60000, + }); + + await rq.start(); + t.after(() => rq.destroy(true).catch(() => undefined)); + mock.method(rq.reader, 'blmove', () => + Promise.reject(Object.assign(new Error('nope'), { code: 'EPIPE' })), + ); + + await rq.readSafe(); + + const lines = matching(cap.warn, /safe reading of queue/); + + assert.equal(lines.length, 1); + assert.match(lines[0], /EPIPE/); + assert.equal(lines[0].includes('nope'), false); + }); + + it('neither rejects nor stays quiet on an error whose message throws', async t => { + const cap = capturing(); + const rq: any = new RedisQueue(uuid(), { + logger: cap.logger, + safeDelivery: true, + safeDeliveryTtl: 60000, + }); + + await rq.start(); + t.after(() => rq.destroy(true).catch(() => undefined)); + + const evil = new Error('boom'); + + Object.defineProperty(evil, 'message', { + get() { + throw new Error('not your business'); + }, + }); + mock.method(rq.reader, 'blmove', () => Promise.reject(evil)); + + await assert.doesNotReject(rq.readSafe()); + assert.equal(matching(cap.warn, /safe reading of queue/).length, 1); + }); + + it('stays quiet when the reader connection was closed on purpose', async t => { + const cap = capturing(); + const rq: any = new RedisQueue(uuid(), { + logger: cap.logger, + safeDelivery: true, + safeDeliveryTtl: 60000, + }); + + await rq.start(); + t.after(() => rq.destroy(true).catch(() => undefined)); + mock.method(rq.reader, 'blmove', () => + Promise.reject(new Error('Stream connection ended')), + ); + + await rq.readSafe(); + + assert.equal(matching(cap.warn, /safe reading of queue/).length, 0); + }); + + it('stays quiet when the reader is gone, as after stop()', async t => { + const cap = capturing(); + const rq: any = new RedisQueue(uuid(), { + logger: cap.logger, + safeDelivery: true, + safeDeliveryTtl: 60000, + }); + + await rq.start(); + t.after(() => rq.destroy(true).catch(() => undefined)); + mock.method(rq.reader, 'blmove', () => { + delete rq.reader; + + return Promise.reject(new Error('read failed')); + }); + + await rq.readSafe(); + + assert.equal(matching(cap.warn, /safe reading of queue/).length, 0); + }); + + it('warns with queue and code when a worker key is not deleted', async t => { + const cap = capturing(); + const rq: any = new RedisQueue(uuid(), { + logger: cap.logger, + safeDelivery: true, + safeDeliveryTtl: 60000, + }); + + await rq.start(); + t.after(() => rq.destroy(true).catch(() => undefined)); + + let popped = 0; + + mock.method(rq.reader, 'blmove', () => { + if (popped++) { + delete rq.reader; + + return Promise.resolve(null); + } + + return Promise.resolve( + pack({ id: uuid(), message: { a: 1 }, from: 'Sender' }), + ); + }); + mock.method(rq.writer, 'del', () => + Promise.reject(new Error('WRONGTYPE nope')), + ); + + await rq.readSafe(); + await tick(); + + const lines = matching(cap.warn, /OnReadSafe: del error/); + + assert.equal(lines.length, 1); + assert.match(lines[0], /WRONGTYPE/); + assert.equal(/worker/.test(lines[0]), false); + }); +}); + +describe('RedisQueue watcher check logging', () => { + afterEach(() => mock.restoreAll()); + + it('warns when the watcher existence check itself fails', async t => { + const cap = capturing(); + const rq: any = new RedisQueue(uuid(), { logger: cap.logger }); + + await rq.start(); + t.after(() => rq.destroy(true).catch(() => undefined)); + mock.method(rq, 'watcherCount', () => + Promise.reject(new Error('LOADING redis is loading')), + ); + + await rq.runWatcherCheck(); + await rq.runWatcherCheck(); + + const lines = matching(cap.warn, /watcher check failed/); + + // one line per tick: the pace is bounded by watcherCheckDelay + assert.equal(lines.length, 2); + assert.match(lines[0], /LOADING/); + assert.equal(rq.watcherCheckBusy, false); + }); + + it('does not duplicate the line watcher initialization writes itself', async t => { + const cap = capturing(); + const rq: any = new RedisQueue(uuid(), { logger: cap.logger }); + + await rq.start(); + t.after(() => rq.destroy(true).catch(() => undefined)); + // the real initWatcher() runs and fails inside, so its own + // unconditional line is the one under test here + mock.method(rq, 'watcherCount', () => Promise.resolve(0)); + mock.method(rq, 'lock', () => + Promise.reject(new Error('LOADING redis is loading')), + ); + + await rq.runWatcherCheck(); + + assert.equal( + matching(cap.error, /error initializing watcher/).length, + 1, + 'watcher initialization must keep reporting its own failure', + ); + assert.equal(matching(cap.warn, /watcher check failed/).length, 0); + }); +}); + +describe('RedisQueue subscription lifecycle logging', () => { + afterEach(() => mock.restoreAll()); + + it('reports a subscription and its restoration unconditionally', async t => { + const cap = capturing(); + const rq: any = new RedisQueue(uuid(), { logger: cap.logger }); + + t.after(() => rq.destroy().catch(() => undefined)); + + await rq.subscribe('FlowEvents', () => undefined); + + assert.equal( + matching(cap.info, /subscribed to channel FlowEvents/).length, + 1, + ); + + await rq.restoreSubscription(); + + assert.equal( + matching(cap.info, /restored subscription to channel FlowEvents/) + .length, + 1, + ); + }); + + it('warns on every failed reconnection attempt', async t => { + const cap = capturing(); + const rq: any = new RedisQueue(uuid(), { logger: cap.logger }); + + t.after(() => rq.destroy().catch(() => undefined)); + mock.method(rq, 'connect', () => + Promise.reject( + Object.assign(new Error('down'), { + code: 'ECONNREFUSED', + }), + ), + ); + mock.method(rq, 'scheduleReconnect', () => undefined); + + await rq.reconnectNow('reader'); + await rq.reconnectNow('reader'); + + const lines = matching(cap.warn, /reconnect of the reader channel/); + + // one line per attempt: the retry pace itself is bounded by the + // reconnection backoff, so no aggregation is applied here + assert.equal(lines.length, 2); + assert.match(lines[0], /ECONNREFUSED/); + }); +}); + +describe('RedisQueue maintenance shutdown logging', () => { + afterEach(() => mock.restoreAll()); + + it('warns before disabling safe-delivery maintenance for good', async t => { + const cap = capturing(); + const rq: any = new RedisQueue(uuid(), { + logger: cap.logger, + host: '127.0.0.99', + port: 6399, + safeDelivery: true, + cleanup: true, + }); + + t.after(() => rq.destroy().catch(() => undefined)); + + await rq.runSafeCheck(); + + const lines = matching(cap.warn, /safe delivery maintenance stopped/); + + assert.equal(lines.length, 1); + assert.match(lines[0], /safeDelivery true/); + assert.match(lines[0], /cleanup true/); + }); +}); + +describe('RedisQueue duplicate-cause logging', () => { + afterEach(() => mock.restoreAll()); + + it('aggregates expired-lease requeues by queue for one pass', async t => { + const cap = capturing(); + const name = uuid(); + const rq: any = new RedisQueue(name, { + logger: cap.logger, + safeDelivery: true, + }); + + await rq.start(); + t.after(() => rq.destroy(true).catch(() => undefined)); + + const expired = (queue: string, worker: string) => + `imq:${queue}:worker:${worker}:${Date.now() - 60000}`; + const first = expired(name, 'abc-lease'); + const second = expired(name, 'def-lease'); + const other = expired('OtherQueue', 'ghi-lease'); + + QS()[first] = ['SECRET-MESSAGE-BODY']; + QS()[second] = ['SECRET-MESSAGE-BODY-TOO']; + QS()[other] = ['SECRET-MESSAGE-BODY-THREE']; + + await rq.processKeys([first, second, other], Date.now()); + + const lines = matching(cap.warn, /re-queued/); + + // one line per destination queue for the whole pass, with a count + assert.equal(lines.length, 2); + + const own = lines.find(one => one.includes(`queue ${name}`)); + const foreign = lines.find(one => one.includes('queue OtherQueue')); + + assert.match(own || '', /re-queued 2 messages/); + assert.match(foreign || '', /re-queued 1 messages/); + + for (const line of lines) { + assert.equal(line.includes('SECRET-MESSAGE-BODY'), false); + assert.equal(line.includes('-lease'), false); + assert.equal(line.includes(':worker:'), false); + assert.equal(line.includes('imq:'), false); + } + }); + + it('still reports requeues done before a move of the pass throws', async t => { + const cap = capturing(); + const name = uuid(); + const rq: any = new RedisQueue(name, { + logger: cap.logger, + safeDelivery: true, + }); + + await rq.start(); + t.after(() => rq.destroy(true).catch(() => undefined)); + + const expired = (worker: string) => + `imq:${name}:worker:${worker}:${Date.now() - 60000}`; + const first = expired('abc'); + const second = expired('def'); + + QS()[first] = ['MSG']; + + let calls = 0; + + mock.method(rq.writer, 'lmove', (key: string) => { + if (++calls === 2) { + return Promise.reject(new Error('LOADING redis is loading')); + } + + return Promise.resolve(QS()[key]?.pop() || null); + }); + + await assert.rejects( + rq.processKeys([first, second], Date.now()), + /LOADING/, + 'the exception must keep escaping exactly as before', + ); + + const lines = matching(cap.warn, /re-queued/); + + assert.equal( + lines.length, + 1, + 'the requeue which did happen must stay reported', + ); + assert.match(lines[0], /re-queued 1 messages/); + }); + + it('says unknown instead of leaking a prefix which carries a colon', async t => { + const cap = capturing(); + const rq: any = new RedisQueue('LeaseColon', { + logger: cap.logger, + prefix: 'tenant:prod', + safeDelivery: true, + }); + + await rq.start(); + t.after(() => rq.destroy(true).catch(() => undefined)); + + // with such a prefix the two-segment arithmetic reconstructs the + // move target as the bare prefix, so the exact-prefix strip fails + const expired = `tenant:prod:LeaseColon:worker:abc:${ + Date.now() - 60000 + }`; + + QS()[expired] = ['MSG']; + + await rq.processKeys([expired], Date.now()); + + const lines = matching(cap.warn, /re-queued/); + + assert.equal(lines.length, 1); + assert.match(lines[0], /to queue unknown/); + assert.equal(lines[0].includes('tenant:prod'), false); + }); + + it('stays quiet when nothing was re-queued', async t => { + const cap = capturing(); + const name = uuid(); + const rq: any = new RedisQueue(name, { + logger: cap.logger, + safeDelivery: true, + }); + + await rq.start(); + t.after(() => rq.destroy(true).catch(() => undefined)); + + const fresh = `imq:${name}:worker:abc:${Date.now() + 60000}`; + + QS()[fresh] = ['MSG']; + + await rq.processKeys([fresh], Date.now()); + + assert.equal(matching(cap.warn, /re-queued/).length, 0); + }); +}); + +describe('RedisQueue cleanup deletion logging', () => { + afterEach(() => mock.restoreAll()); + + it('warns with counts only when keys were really removed', async t => { + const cap = capturing(); + const rq: any = new RedisQueue( + 'CleanLogged', + { logger: cap.logger, cleanup: true, cleanupFilter: '*' }, + IMQMode.PUBLISHER, + ); + + await rq.start(); + t.after(() => rq.destroy().catch(() => undefined)); + + const client = 'imq:GoneLogged:writer:pid:1:host:h'; + + QS()['imq:GoneLogged'] = ['pending']; + CL()[client] = true; + mock.method(rq.writer, 'scan', async () => ['0', ['imq:GoneLogged']]); + + await rq.processCleanup(); + + assert.equal(matching(cap.warn, /cleanup removed/).length, 0); + + delete CL()[client]; + + await rq.processCleanup(); + + assert.equal(matching(cap.warn, /cleanup removed/).length, 0); + + await rq.processCleanup(); + + const lines = matching(cap.warn, /cleanup removed/); + + assert.equal(lines.length, 1); + assert.match(lines[0], /removed 1 of 1 candidate keys/); + assert.equal(lines[0].includes('GoneLogged'), false); + }); +}); + +describe('RedisQueue publish visibility', () => { + afterEach(() => mock.restoreAll()); + + it('warns once when redis reports no subscribers', async t => { + const cap = capturing(); + const rq: any = new RedisQueue( + 'PubNoSubs', + { logger: cap.logger }, + IMQMode.PUBLISHER, + ); + + await rq.start(); + t.after(() => rq.destroy().catch(() => undefined)); + mock.method(rq.writer, 'publish', () => 0); + + await rq.publish({ ssn: '000-00-0000' }); + await rq.publish({ ssn: '000-00-0000' }); + + const lines = matching(cap.warn, /no subscribers/); + + assert.equal(lines.length, 1); + assert.match(lines[0], /PubNoSubs/); + assert.equal(lines[0].includes('000-00-0000'), false); + }); + + it('keeps the state of every channel apart', async t => { + const cap = capturing(); + const rq: any = new RedisQueue( + 'PubTwoChannels', + { logger: cap.logger }, + IMQMode.PUBLISHER, + ); + + await rq.start(); + t.after(() => rq.destroy().catch(() => undefined)); + mock.method(rq.writer, 'publish', () => 0); + + await rq.publish({ a: 1 }, 'ChannelA'); + await rq.publish({ a: 1 }, 'ChannelB'); + await rq.publish({ a: 1 }, 'ChannelA'); + await rq.publish({ a: 1 }, 'ChannelB'); + + assert.equal(matching(cap.warn, /no subscribers/).length, 2); + }); + + it('keeps the transition state bounded, reporting the rest every time', async t => { + const cap = capturing(); + const rq: any = new RedisQueue( + 'PubBounded', + { logger: cap.logger }, + IMQMode.PUBLISHER, + ); + + await rq.start(); + t.after(() => rq.destroy().catch(() => undefined)); + mock.method(rq.writer, 'publish', () => 0); + + for (let i = 0; i < 130; i++) { + await rq.publish({ a: 1 }, `Channel-${i}`); + } + + assert.equal( + rq.noSubscribers.size, + 128, + 'channel names above the bound must not be remembered', + ); + + const before = matching(cap.warn, /no subscribers/).length; + + // a channel above the bound is not remembered, so its publishes + // keep being reported - the price of a memory-bounded set + await rq.publish({ a: 1 }, 'Channel-129'); + await rq.publish({ a: 1 }, 'Channel-129'); + + assert.equal(matching(cap.warn, /no subscribers/).length, before + 2); + }); + + it('stays quiet when the reply is not a number', async t => { + const cap = capturing(); + const rq: any = new RedisQueue( + 'PubNoNumber', + { logger: cap.logger }, + IMQMode.PUBLISHER, + ); + + await rq.start(); + t.after(() => rq.destroy().catch(() => undefined)); + mock.method(rq.writer, 'publish', () => undefined); + + await rq.publish({ a: 1 }); + + assert.equal(matching(cap.warn, /no subscribers/).length, 0); + }); + + it('does not reject when the reply resists being read', async t => { + const cap = capturing(); + const rq: any = new RedisQueue( + 'PubEvilReply', + { logger: cap.logger }, + IMQMode.PUBLISHER, + ); + + await rq.start(); + t.after(() => rq.destroy().catch(() => undefined)); + mock.method(rq.writer, 'publish', () => ({ + [Symbol.toPrimitive]() { + throw new Error('not your business'); + }, + })); + + await assert.doesNotReject(rq.publish({ a: 1 })); + assert.equal(matching(cap.warn, /no subscribers/).length, 0); + }); + + it('stays quiet while redis reports subscribers', async t => { + const cap = capturing(); + const rq: any = new RedisQueue( + 'PubWithSubs', + { logger: cap.logger }, + IMQMode.PUBLISHER, + ); + + await rq.start(); + t.after(() => rq.destroy().catch(() => undefined)); + + await rq.publish({ a: 1 }); + + assert.equal(matching(cap.warn, /no subscribers/).length, 0); + }); +}); diff --git a/test/unit/helpers/logging.spec.ts b/test/unit/helpers/logging.spec.ts new file mode 100644 index 0000000..b7edd2d --- /dev/null +++ b/test/unit/helpers/logging.spec.ts @@ -0,0 +1,109 @@ +/*! + * I'm Queue Software Project + * Copyright (C) 2025 imqueue.com + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + * + * If you want to use this code in a closed source (commercial) project, you can + * purchase a proprietary commercial license. Please contact us at + * to get commercial licensing options. + */ +import { describe, it } from 'node:test'; +import assert from 'node:assert/strict'; +import { errorCode } from '../../../src/helpers/index.js'; + +describe('errorCode()', () => { + it('prefers an explicit string code', () => { + assert.equal( + errorCode( + Object.assign(new Error('nope'), { code: 'ECONNREFUSED' }), + ), + 'ECONNREFUSED', + ); + }); + + it('accepts a numeric code', () => { + assert.equal(errorCode({ code: 42 }), '42'); + }); + + it('reads the leading redis reply code', () => { + assert.equal( + errorCode(new Error('WRONGTYPE Operation against a key')), + 'WRONGTYPE', + ); + }); + + it('accepts a framework code', () => { + assert.equal( + errorCode({ code: 'IMQ_RPC_CALL_TIMEOUT' }), + 'IMQ_RPC_CALL_TIMEOUT', + ); + }); + + it('maps a known client failure message to a code', () => { + assert.equal( + errorCode(new Error('Connection is closed.')), + 'CONNECTION_CLOSED', + ); + assert.equal( + errorCode(new Error('Stream connection ended by server')), + 'STREAM_ENDED', + ); + }); + + it('never returns the message itself', () => { + assert.equal( + errorCode(new Error('user 12345 phone +100000000')), + 'unknown', + ); + }); + + it('never returns an upper-case word of an unknown message', () => { + assert.equal(errorCode(new Error('CUSTOMER 12345 secret')), 'unknown'); + }); + + it('never returns a code outside the allow-list', () => { + assert.equal(errorCode({ code: 'SSN-000-00-0000' }), 'unknown'); + assert.equal( + errorCode({ code: 'CUSTOMER_12345_NOT_FOUND' }), + 'unknown', + ); + assert.equal(errorCode({ code: 123456789 }), 'unknown'); + assert.equal(errorCode({ code: -1 }), 'unknown'); + assert.equal(errorCode({ code: 1.5 }), 'unknown'); + }); + + it('never returns the error class name', () => { + assert.equal(errorCode(new TypeError('boom')), 'unknown'); + assert.equal(errorCode({ name: 'Customer_12345' }), 'unknown'); + }); + + it('never throws on odd values', () => { + assert.equal(errorCode(undefined), 'unknown'); + assert.equal(errorCode(null), 'unknown'); + assert.equal(errorCode('a string'), 'unknown'); + assert.equal(errorCode(0), 'unknown'); + assert.equal(errorCode({}), 'unknown'); + }); + + it('never throws when reading a property throws', () => { + const evil = { + get code(): string { + throw new Error('nope'); + }, + }; + + assert.equal(errorCode(evil), 'unknown'); + }); +});