feat(logging): report transport failures that were silent or verbose-only

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

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<void> {
         const promises: Array<Promise<void>> = [];
@@ -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;
+            }
         }
     }