redis/node-redis
1
0
Fork 0
mirror of https://github.com/redis/node-redis.git synced 2025-12-11 09:22:35 +03:00
Code Activity

fix(init): option parsing (#3086)

Browse Source 
* refactor(maint): rename options

* fix(init): option parsing

* refactor(client): #options cannot be undefined
This commit is contained in:
Nikolay Karadzhov
2025-09-23 14:24:37 +03:00
committed by GitHub
parent ed7b905479
commit e138cbd05b
4 changed files with 108 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
packages/client/lib/client/enterprise-maintenance-manager.spec.ts Normal file
View File

@@ -0,0 +1,49 @@
import assert from "node:assert";
import { createClient } from "../../";
describe("EnterpriseMaintenanceManager does not prevent proper options parsing", () => {
it("should not throw when initializing without options", async () => {
const client = createClient();
assert.doesNotThrow(async () => {
//Expected to reject because there is no url or socket provided and there is no running server on localhost
await assert.rejects(client.connect);
});
});
it("should not throw when initializing without url/socket and with maint", async () => {
const client = createClient({
maintNotifications: "enabled",
RESP: 3,
});
assert.doesNotThrow(async () => {
//Expected to reject because there is no url or socket provided and there is no running server on localhost
await assert.rejects(client.connect);
});
});
it("should not throw when initializing with url and with maint", async () => {
const client = createClient({
maintNotifications: "enabled",
RESP: 3,
url: "redis://localhost:6379",
});
assert.doesNotThrow(async () => {
//Expected to reject because there is no url or socket provided and there is no running server on localhost
await assert.rejects(client.connect);
});
});
it("should not throw when initializing with socket and with maint", async () => {
const client = createClient({
maintNotifications: "enabled",
RESP: 3,
socket: {
host: "localhost",
port: 6379,
},
});
assert.doesNotThrow(async () => {
//Expected to reject because there is no url or socket provided and there is no running server on localhost
await assert.rejects(client.connect);
});
});
});

34
packages/client/lib/client/enterprise-maintenance-manager.ts
View File

@@ -5,7 +5,7 @@ import { isIP } from "net";
import { lookup } from "dns/promises"; import { lookup } from "dns/promises";
import assert from "node:assert"; import assert from "node:assert";
import { setTimeout } from "node:timers/promises"; import { setTimeout } from "node:timers/promises";
import RedisSocket from "./socket"; import RedisSocket, { RedisTcpSocketOptions } from "./socket";
import diagnostics_channel from "node:diagnostics_channel"; import diagnostics_channel from "node:diagnostics_channel";
export const MAINTENANCE_EVENTS = { export const MAINTENANCE_EVENTS = {
@@ -64,12 +64,12 @@ export default class EnterpriseMaintenanceManager {
#client: Client; #client: Client;
static setupDefaultMaintOptions(options: RedisClientOptions) { static setupDefaultMaintOptions(options: RedisClientOptions) {
if (options.maintPushNotifications === undefined) { if (options.maintNotifications === undefined) {
options.maintPushNotifications = options.maintNotifications =
options?.RESP === 3 ? "auto" : "disabled"; options?.RESP === 3 ? "auto" : "disabled";
} }
if (options.maintMovingEndpointType === undefined) { if (options.maintEndpointType === undefined) {
options.maintMovingEndpointType = "auto"; options.maintEndpointType = "auto";
} }
if (options.maintRelaxedSocketTimeout === undefined) { if (options.maintRelaxedSocketTimeout === undefined) {
options.maintRelaxedSocketTimeout = 10000; options.maintRelaxedSocketTimeout = 10000;
@@ -80,14 +80,20 @@ export default class EnterpriseMaintenanceManager {
} }
static async getHandshakeCommand( static async getHandshakeCommand(
tls: boolean,
host: string,
options: RedisClientOptions, options: RedisClientOptions,
): Promise< ): Promise<
| { cmd: Array<RedisArgument>; errorHandler: (error: Error) => void } | { cmd: Array<RedisArgument>; errorHandler: (error: Error) => void }
| undefined | undefined
> { > {
if (options.maintPushNotifications === "disabled") return; if (options.maintNotifications === "disabled") return;
const host = options.url
? new URL(options.url).hostname
: (options.socket as RedisTcpSocketOptions | undefined)?.host;
if (!host) return;
const tls = options.socket?.tls ?? false
const movingEndpointType = await determineEndpoint(tls, host, options); const movingEndpointType = await determineEndpoint(tls, host, options);
return { return {
@@ -100,7 +106,7 @@ export default class EnterpriseMaintenanceManager {
], ],
errorHandler: (error: Error) => { errorHandler: (error: Error) => {
dbgMaintenance("handshake failed:", error); dbgMaintenance("handshake failed:", error);
if (options.maintPushNotifications === "enabled") { if (options.maintNotifications === "enabled") {
throw error; throw error;
} }
}, },
@@ -189,7 +195,7 @@ export default class EnterpriseMaintenanceManager {
// reconnect to its currently configured endpoint after half of the grace // reconnect to its currently configured endpoint after half of the grace
// period that was communicated by the server is over. // period that was communicated by the server is over.
if (url === null) { if (url === null) {
assert(this.#options.maintMovingEndpointType === "none"); assert(this.#options.maintEndpointType === "none");
assert(this.#options.socket !== undefined); assert(this.#options.socket !== undefined);
assert("host" in this.#options.socket); assert("host" in this.#options.socket);
assert(typeof this.#options.socket.host === "string"); assert(typeof this.#options.socket.host === "string");
@@ -329,12 +335,12 @@ async function determineEndpoint(
host: string, host: string,
options: RedisClientOptions, options: RedisClientOptions,
): Promise<MovingEndpointType> { ): Promise<MovingEndpointType> {
assert(options.maintMovingEndpointType !== undefined); assert(options.maintEndpointType !== undefined);
if (options.maintMovingEndpointType !== "auto") { if (options.maintEndpointType !== "auto") {
dbgMaintenance( dbgMaintenance(
`Determine endpoint type: ${options.maintMovingEndpointType}`, `Determine endpoint type: ${options.maintEndpointType}`,
); );
return options.maintMovingEndpointType; return options.maintEndpointType;
} }
const ip = isIP(host) ? host : (await lookup(host, { family: 0 })).address; const ip = isIP(host) ? host : (await lookup(host, { family: 0 })).address;

2
packages/client/lib/client/index.spec.ts
View File

@@ -1,7 +1,7 @@
import { strict as assert } from 'node:assert'; import { strict as assert } from 'node:assert';
import testUtils, { GLOBAL, waitTillBeenCalled } from '../test-utils'; import testUtils, { GLOBAL, waitTillBeenCalled } from '../test-utils';
import RedisClient, { RedisClientOptions, RedisClientType } from '.'; import RedisClient, { RedisClientOptions, RedisClientType } from '.';
import { AbortError, ClientClosedError, ClientOfflineError, ConnectionTimeoutError, DisconnectsClientError, ErrorReply, MultiErrorReply, SocketClosedUnexpectedlyError, TimeoutError, WatchError } from '../errors'; import { AbortError, ClientClosedError, ClientOfflineError, ConnectionTimeoutError, DisconnectsClientError, ErrorReply, MultiErrorReply, TimeoutError, WatchError } from '../errors';
import { defineScript } from '../lua-script'; import { defineScript } from '../lua-script';
import { spy, stub } from 'sinon'; import { spy, stub } from 'sinon';
import { once } from 'node:events'; import { once } from 'node:events';

77
packages/client/lib/client/index.ts
View File

@@ -1,5 +1,5 @@
import COMMANDS from '../commands'; import COMMANDS from '../commands';
import RedisSocket, { RedisSocketOptions, RedisTcpSocketOptions } from './socket'; import RedisSocket, { RedisSocketOptions } from './socket';
import { BasicAuth, CredentialsError, CredentialsProvider, StreamingCredentialsProvider, UnableToObtainNewCredentialsError, Disposable } from '../authx'; import { BasicAuth, CredentialsError, CredentialsProvider, StreamingCredentialsProvider, UnableToObtainNewCredentialsError, Disposable } from '../authx';
import RedisCommandsQueue, { CommandOptions } from './commands-queue'; import RedisCommandsQueue, { CommandOptions } from './commands-queue';
import { EventEmitter } from 'node:events'; import { EventEmitter } from 'node:events';
@@ -154,7 +154,7 @@ export interface RedisClientOptions<
* *
* The default is `auto`. * The default is `auto`.
*/ */
maintPushNotifications?: 'disabled' | 'enabled' | 'auto'; maintNotifications?: 'disabled' | 'enabled' | 'auto';
/** /**
* Controls how the client requests the endpoint to reconnect to during a MOVING notification in Redis Enterprise maintenance. * Controls how the client requests the endpoint to reconnect to during a MOVING notification in Redis Enterprise maintenance.
* *
@@ -167,19 +167,19 @@ export interface RedisClientOptions<
* The default is `auto`. * The default is `auto`.
*/ */
maintMovingEndpointType?: MovingEndpointType; maintEndpointType?: MovingEndpointType;
/** /**
* Specifies a more relaxed timeout (in milliseconds) for commands during a maintenance window. * Specifies a more relaxed timeout (in milliseconds) for commands during a maintenance window.
* This helps minimize command timeouts during maintenance. If not provided, the `commandOptions.timeout` * This helps minimize command timeouts during maintenance. Timeouts during maintenance period result
* will be used instead. Timeouts during maintenance period result in a `CommandTimeoutDuringMaintenance` error. * in a `CommandTimeoutDuringMaintenance` error.
* *
* The default is 10000 * The default is 10000
*/ */
maintRelaxedCommandTimeout?: number; maintRelaxedCommandTimeout?: number;
/** /**
* Specifies a more relaxed timeout (in milliseconds) for the socket during a maintenance window. * Specifies a more relaxed timeout (in milliseconds) for the socket during a maintenance window.
* This helps minimize socket timeouts during maintenance. If not provided, the `socket.timeout` * This helps minimize socket timeouts during maintenance. Timeouts during maintenance period result
* will be used instead. Timeouts during maintenance period result in a `SocketTimeoutDuringMaintenance` error. * in a `SocketTimeoutDuringMaintenance` error.
* *
* The default is 10000 * The default is 10000
*/ */
@@ -429,7 +429,7 @@ export default class RedisClient<
return parsed; return parsed;
} }
readonly #options?: RedisClientOptions<M, F, S, RESP, TYPE_MAPPING>; readonly #options: RedisClientOptions<M, F, S, RESP, TYPE_MAPPING>;
#socket: RedisSocket; #socket: RedisSocket;
readonly #queue: RedisCommandsQueue; readonly #queue: RedisCommandsQueue;
#selectedDB = 0; #selectedDB = 0;
@@ -453,7 +453,7 @@ export default class RedisClient<
return this._self.#clientSideCache; return this._self.#clientSideCache;
} }
get options(): RedisClientOptions<M, F, S, RESP> | undefined { get options(): RedisClientOptions<M, F, S, RESP> {
return this._self.#options; return this._self.#options;
} }
@@ -503,15 +503,15 @@ export default class RedisClient<
this.#socket = this.#initiateSocket(); this.#socket = this.#initiateSocket();
if(options?.maintPushNotifications !== 'disabled') { if(this.#options.maintNotifications !== 'disabled') {
new EnterpriseMaintenanceManager(this.#queue, this, this.#options!); new EnterpriseMaintenanceManager(this.#queue, this, this.#options);
}; };
if (options?.clientSideCache) { if (this.#options.clientSideCache) {
if (options.clientSideCache instanceof ClientSideCacheProvider) { if (this.#options.clientSideCache instanceof ClientSideCacheProvider) {
this.#clientSideCache = options.clientSideCache; this.#clientSideCache = this.#options.clientSideCache;
} else { } else {
const cscConfig = options.clientSideCache; const cscConfig = this.#options.clientSideCache;
this.#clientSideCache = new BasicClientSideCache(cscConfig); this.#clientSideCache = new BasicClientSideCache(cscConfig);
} }
this.#queue.addPushHandler((push: Array<any>): boolean => { this.#queue.addPushHandler((push: Array<any>): boolean => {
@@ -535,16 +535,16 @@ export default class RedisClient<
throw new Error('Client Side Caching is only supported with RESP3'); throw new Error('Client Side Caching is only supported with RESP3');
} }
if (options?.maintPushNotifications && options?.maintPushNotifications !== 'disabled' && options?.RESP !== 3) { if (options?.maintNotifications && options?.maintNotifications !== 'disabled' && options?.RESP !== 3) {
throw new Error('Graceful Maintenance is only supported with RESP3'); throw new Error('Graceful Maintenance is only supported with RESP3');
} }
} }
#initiateOptions(options?: RedisClientOptions<M, F, S, RESP, TYPE_MAPPING>): RedisClientOptions<M, F, S, RESP, TYPE_MAPPING> | undefined { #initiateOptions(options: RedisClientOptions<M, F, S, RESP, TYPE_MAPPING> = {}): RedisClientOptions<M, F, S, RESP, TYPE_MAPPING> {
// Convert username/password to credentialsProvider if no credentialsProvider is already in place // Convert username/password to credentialsProvider if no credentialsProvider is already in place
if (!options?.credentialsProvider && (options?.username || options?.password)) { if (!options.credentialsProvider && (options.username || options.password)) {
options.credentialsProvider = { options.credentialsProvider = {
type: 'async-credentials-provider', type: 'async-credentials-provider',
@@ -555,19 +555,19 @@ export default class RedisClient<
}; };
} }
if (options?.database) { if (options.database) {
this._self.#selectedDB = options.database; this._self.#selectedDB = options.database;
} }
if (options?.commandOptions) { if (options.commandOptions) {
this._commandOptions = options.commandOptions; this._commandOptions = options.commandOptions;
} }
if(options?.maintPushNotifications !== 'disabled') { if(options.maintNotifications !== 'disabled') {
EnterpriseMaintenanceManager.setupDefaultMaintOptions(options!); EnterpriseMaintenanceManager.setupDefaultMaintOptions(options);
} }
if (options?.url) { if (options.url) {
const parsedOptions = RedisClient.parseOptions(options); const parsedOptions = RedisClient.parseOptions(options);
if (parsedOptions?.database) { if (parsedOptions?.database) {
this._self.#selectedDB = parsedOptions.database; this._self.#selectedDB = parsedOptions.database;
@@ -580,8 +580,8 @@ export default class RedisClient<
#initiateQueue(): RedisCommandsQueue { #initiateQueue(): RedisCommandsQueue {
return new RedisCommandsQueue( return new RedisCommandsQueue(
this.#options?.RESP ?? 2, this.#options.RESP ?? 2,
this.#options?.commandsQueueMaxLength, this.#options.commandsQueueMaxLength,
(channel, listeners) => this.emit('sharded-channel-moved', channel, listeners) (channel, listeners) => this.emit('sharded-channel-moved', channel, listeners)
); );
} }
@@ -591,7 +591,7 @@ export default class RedisClient<
*/ */
private reAuthenticate = async (credentials: BasicAuth) => { private reAuthenticate = async (credentials: BasicAuth) => {
// Re-authentication is not supported on RESP2 with PubSub active // Re-authentication is not supported on RESP2 with PubSub active
if (!(this.isPubSubActive && !this.#options?.RESP)) { if (!(this.isPubSubActive && !this.#options.RESP)) {
await this.sendCommand( await this.sendCommand(
parseArgs(COMMANDS.AUTH, { parseArgs(COMMANDS.AUTH, {
username: credentials.username, username: credentials.username,
@@ -640,9 +640,9 @@ export default class RedisClient<
Array<{ cmd: CommandArguments } & { errorHandler?: (err: Error) => void }> Array<{ cmd: CommandArguments } & { errorHandler?: (err: Error) => void }>
> { > {
const commands = []; const commands = [];
const cp = this.#options?.credentialsProvider; const cp = this.#options.credentialsProvider;
if (this.#options?.RESP) { if (this.#options.RESP) {
const hello: HelloOptions = {}; const hello: HelloOptions = {};
if (cp && cp.type === 'async-credentials-provider') { if (cp && cp.type === 'async-credentials-provider') {
@@ -702,7 +702,7 @@ export default class RedisClient<
} }
} }
if (this.#options?.name) { if (this.#options.name) {
commands.push({ commands.push({
cmd: parseArgs(COMMANDS.CLIENT_SETNAME, this.#options.name) cmd: parseArgs(COMMANDS.CLIENT_SETNAME, this.#options.name)
}); });
@@ -713,11 +713,11 @@ export default class RedisClient<
commands.push({ cmd: ['SELECT', this.#selectedDB.toString()] }); commands.push({ cmd: ['SELECT', this.#selectedDB.toString()] });
} }
if (this.#options?.readonly) { if (this.#options.readonly) {
commands.push({ cmd: parseArgs(COMMANDS.READONLY) }); commands.push({ cmd: parseArgs(COMMANDS.READONLY) });
} }
if (!this.#options?.disableClientInfo) { if (!this.#options.disableClientInfo) {
commands.push({ commands.push({
cmd: ['CLIENT', 'SETINFO', 'LIB-VER', version], cmd: ['CLIENT', 'SETINFO', 'LIB-VER', version],
errorHandler: () => { errorHandler: () => {
@@ -732,7 +732,7 @@ export default class RedisClient<
'CLIENT', 'CLIENT',
'SETINFO', 'SETINFO',
'LIB-NAME', 'LIB-NAME',
this.#options?.clientInfoTag this.#options.clientInfoTag
? `node-redis(${this.#options.clientInfoTag})` ? `node-redis(${this.#options.clientInfoTag})`
: 'node-redis' : 'node-redis'
], ],
@@ -748,8 +748,7 @@ export default class RedisClient<
commands.push({cmd: this.#clientSideCache.trackingOn()}); commands.push({cmd: this.#clientSideCache.trackingOn()});
} }
const { tls, host } = this.#options!.socket as RedisTcpSocketOptions; const maintenanceHandshakeCmd = await EnterpriseMaintenanceManager.getHandshakeCommand(this.#options);
const maintenanceHandshakeCmd = await EnterpriseMaintenanceManager.getHandshakeCommand(!!tls, host!, this.#options!);
if(maintenanceHandshakeCmd) { if(maintenanceHandshakeCmd) {
commands.push(maintenanceHandshakeCmd); commands.push(maintenanceHandshakeCmd);
}; };
@@ -769,7 +768,7 @@ export default class RedisClient<
.on('error', err => { .on('error', err => {
this.emit('error', err); this.emit('error', err);
this.#clientSideCache?.onError(); this.#clientSideCache?.onError();
if (this.#socket.isOpen && !this.#options?.disableOfflineQueue) { if (this.#socket.isOpen && !this.#options.disableOfflineQueue) {
this.#queue.flushWaitingForReply(err); this.#queue.flushWaitingForReply(err);
} else { } else {
this.#queue.flushAll(err); this.#queue.flushAll(err);
@@ -817,7 +816,7 @@ export default class RedisClient<
} }
}; };
const socket = new RedisSocket(socketInitiator, this.#options?.socket); const socket = new RedisSocket(socketInitiator, this.#options.socket);
this.#attachListeners(socket); this.#attachListeners(socket);
return socket; return socket;
} }
@@ -825,7 +824,7 @@ export default class RedisClient<
#pingTimer?: NodeJS.Timeout; #pingTimer?: NodeJS.Timeout;
#setPingTimer(): void { #setPingTimer(): void {
if (!this.#options?.pingInterval || !this.#socket.isReady) return; if (!this.#options.pingInterval || !this.#socket.isReady) return;
clearTimeout(this.#pingTimer); clearTimeout(this.#pingTimer);
this.#pingTimer = setTimeout(() => { this.#pingTimer = setTimeout(() => {
@@ -986,7 +985,7 @@ export default class RedisClient<
transformReply: TransformReply | undefined, transformReply: TransformReply | undefined,
) { ) {
const csc = this._self.#clientSideCache; const csc = this._self.#clientSideCache;
const defaultTypeMapping = this._self.#options?.commandOptions === commandOptions; const defaultTypeMapping = this._self.#options.commandOptions === commandOptions;
const fn = () => { return this.sendCommand(parser.redisArgs, commandOptions) }; const fn = () => { return this.sendCommand(parser.redisArgs, commandOptions) };
@@ -1035,7 +1034,7 @@ export default class RedisClient<
): Promise<T> { ): Promise<T> {
if (!this._self.#socket.isOpen) { if (!this._self.#socket.isOpen) {
return Promise.reject(new ClientClosedError()); return Promise.reject(new ClientClosedError());
} else if (!this._self.#socket.isReady && this._self.#options?.disableOfflineQueue) { } else if (!this._self.#socket.isReady && this._self.#options.disableOfflineQueue) {
return Promise.reject(new ClientOfflineError()); return Promise.reject(new ClientOfflineError());
} }