add debug fn

Author: nkaradzhov

packages/client/lib/client/commands-queue.ts

@@ -5,6 +5,7 @@ import { TypeMapping, ReplyUnion, RespVersions, RedisArgument } from '../RESP/ty
 import { ChannelListeners, PubSub, PubSubCommand, PubSubListener, PubSubType, PubSubTypeListeners } from './pub-sub';
 import { AbortError, ErrorReply, CommandTimeoutDuringMaintananceError, TimeoutError } from '../errors';
 import { MonitorCallback } from '.';
+import { dbgMaintenance } from './enterprise-maintenance-manager';
 
 export interface CommandOptions<T = TypeMapping> {
   chainId?: symbol;
@@ -79,11 +80,14 @@ export default class RedisCommandsQueue {
   #maintenanceCommandTimeout: number | undefined
 
   setMaintenanceCommandTimeout(ms: number | undefined) {
+    dbgMaintenance(`Setting maintenance command timeout to ${ms}`);
     // Prevent possible api misuse
     if (this.#maintenanceCommandTimeout === ms) return;
 
     this.#maintenanceCommandTimeout = ms;
 
+    let counter = 0;
+
     // Overwrite timeouts of all eligible toWrite commands
     this.#toWrite.forEachNode(node => {
       const command = node.value;
@@ -96,6 +100,7 @@ export default class RedisCommandsQueue {
       // if no timeout is given and the command didnt have any timeout before, skip
       if (!newTimeout)  return;
 
+      counter++;
 
       // Overwrite the command's timeout
       const signal = AbortSignal.timeout(newTimeout);
@@ -109,6 +114,7 @@ export default class RedisCommandsQueue {
       };
       signal.addEventListener('abort', command.timeout.listener, { once: true });
     });
+    dbgMaintenance(`Total of ${counter} timeouts reset to ${ms}`);
   }
 
   get isPubSubActive() {

packages/client/lib/client/enterprise-maintenance-manager.ts

@@ -1,7 +1,7 @@
 import EventEmitter from "events";
 import { RedisClientOptions } from ".";
 import RedisCommandsQueue from "./commands-queue";
-import RedisSocket, { RedisSocketOptions, RedisTcpSocketOptions } from "./socket";
+import RedisSocket from "./socket";
 import { RedisArgument } from "../..";
 import { isIP } from "net";
 import { lookup } from "dns/promises";
@@ -25,6 +25,11 @@ export interface SocketTimeoutUpdate {
   timeout?: number;
 }
 
+export const dbgMaintenance = (...args: any[]) => {
+  if (!process.env.DEBUG_MAINTENANCE) return;
+  return console.log('[MNT]', ...args);
+}
+
 export default class EnterpriseMaintenanceManager extends EventEmitter {
   #commandsQueue: RedisCommandsQueue;
   #options: RedisClientOptions;
@@ -47,16 +52,19 @@ export default class EnterpriseMaintenanceManager extends EventEmitter {
       case PN.MOVING: {
         const [_, afterMs, url] = push;
         const [host, port] = url.toString().split(":");
+        dbgMaintenance('Received MOVING:', afterMs, host, Number(port));
         this.#onMoving(afterMs, host, Number(port));
         return true;
       }
       case PN.MIGRATING:
       case PN.FAILING_OVER: {
+        dbgMaintenance('Received MIGRATING|FAILING_OVER');
         this.#onMigrating();
         return true;
       }
       case PN.MIGRATED:
       case PN.FAILED_OVER: {
+        dbgMaintenance('Received MIGRATED|FAILED_OVER');
         this.#onMigrated();
         return true;
       }
@@ -83,6 +91,7 @@ export default class EnterpriseMaintenanceManager extends EventEmitter {
   ): Promise<void> => {
     // 1 [EVENT] MOVING PN received
     // 2 [ACTION] Pause writing
+    dbgMaintenance('Pausing writing of new commands to old socket');
     this.emit(MAINTENANCE_EVENTS.PAUSE_WRITING);
     this.#onMigrating();
 
@@ -93,13 +102,18 @@ export default class EnterpriseMaintenanceManager extends EventEmitter {
     });
     //todo
     newSocket.setMaintenanceTimeout();
+    dbgMaintenance(`Connecting to new socket: ${host}:${port}`);
     await newSocket.connect();
+    dbgMaintenance(`Connected to new socket`);
     // 3 [EVENT] New socket connected
 
+    dbgMaintenance(`Wait for all in-flight commands to complete`);
     await this.#commandsQueue.waitForInflightCommandsToComplete();
+    dbgMaintenance(`In-flight commands completed`);
     // 4 [EVENT] In-flight commands completed
 
     // 5 + 6
+    dbgMaintenance('Resume writing')
     this.emit(MAINTENANCE_EVENTS.RESUME_WRITING, newSocket);
     this.#onMigrated();
   };

packages/client/lib/client/index.ts

@@ -20,7 +20,7 @@ import { BasicClientSideCache, ClientSideCacheConfig, ClientSideCacheProvider }
 import { BasicCommandParser, CommandParser } from './parser';
 import SingleEntryCache from '../single-entry-cache';
 import { version } from '../../package.json'
-import EnterpriseMaintenanceManager, { MAINTENANCE_EVENTS, SocketTimeoutUpdate } from './enterprise-maintenance-manager';
+import EnterpriseMaintenanceManager, { MAINTENANCE_EVENTS, SocketTimeoutUpdate, dbgMaintenance } from './enterprise-maintenance-manager';
 import assert from 'node:assert';
 
 export interface RedisClientOptions<
@@ -752,7 +752,7 @@ export default class RedisClient<
       commands.push({
         cmd: await EnterpriseMaintenanceManager.getHandshakeCommand(tls, host),
         errorHandler: (err: Error) => {
-          console.log("Maintenance handshake failed: ", err);
+          dbgMaintenance("Maintenance handshake failed: ", err);
         }
       });
     }

packages/client/lib/client/socket.ts

@@ -4,6 +4,7 @@ import tls from 'node:tls';
 import { ConnectionTimeoutError, ClientClosedError, SocketClosedUnexpectedlyError, ReconnectStrategyError, SocketTimeoutError, SocketTimeoutDuringMaintananceError } from '../errors';
 import { setTimeout } from 'node:timers/promises';
 import { RedisArgument } from '../RESP/types';
+import { dbgMaintenance } from './enterprise-maintenance-manager';
 
 type NetOptions = {
   tls?: false;
@@ -243,6 +244,7 @@ export default class RedisSocket extends EventEmitter {
   }
 
   setMaintenanceTimeout(ms?: number) {
+    dbgMaintenance(`Reset socket timeout to ${ms}`);
     if (this.#maintenanceTimeout === ms) return;
 
     this.#maintenanceTimeout = ms;