7298903038

Committed 22 Dec 2023 10:32AM UTC coverage: 89.571% (-0.08%) from 89.65%

Build # 7298903038

Build Type

Pull #2350

github

Committed by

web-flow

Commit Message

Merge 1f64edfac into fb0d8a7ac

Pull Request Pull Request #2350: fix(worker): fix so that worker can be closed if Redis is down

Run Details

1046 of 1252 branches covered (0.0%)

Branch coverage included in aggregate %.

10 of 11 new or added lines in 2 files covered. (90.91%)

1 existing line in 1 file now uncovered.

2089 of 2248 relevant lines covered (92.93%)

2979.07 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

86.8

/src/classes/redis-connection.ts

import { EventEmitter } from 'events';
import { default as IORedis } from 'ioredis';
// eslint-disable-next-line @typescript-eslint/ban-ts-comment
// @ts-ignore
import { CONNECTION_CLOSED_ERROR_MSG } from 'ioredis/built/utils';
import { ConnectionOptions, RedisOptions, RedisClient } from '../interfaces';
import {
  decreaseMaxListeners,
  increaseMaxListeners,
  isNotConnectionError,
  isRedisCluster,
  isRedisInstance,
  isRedisVersionLowerThan,
} from '../utils';
import * as scripts from '../scripts';

const overrideMessage = [
  'BullMQ: WARNING! Your redis options maxRetriesPerRequest must be null',
  'and will be overridden by BullMQ.',
].join(' ');

const deprecationMessage =
  'BullMQ: Your redis options maxRetriesPerRequest must be null.';

interface RedisCapabilities {
  canDoubleTimeout: boolean;
}

export interface RawCommand {
  content: string;
  name: string;
  keys: number;
}

export class RedisConnection extends EventEmitter {
  static minimumVersion = '5.0.0';
  static recommendedMinimumVersion = '6.2.0';

  closing: boolean;
  capabilities: RedisCapabilities = {
    canDoubleTimeout: false,
  };

  status: 'initializing' | 'ready' | 'closing' | 'closed' = 'initializing';

  protected _client: RedisClient;

  private readonly opts: RedisOptions;
  private readonly initializing: Promise<RedisClient>;

  private version: string;
  private skipVersionCheck: boolean;
  private handleClientError: (e: Error) => void;
  private handleClientClose: () => void;
  private handleClientReady: () => void;

  constructor(
    opts?: ConnectionOptions,
    private readonly shared: boolean = false,
    private readonly blocking = true,
    skipVersionCheck = false,
  ) {
    super();

    if (!isRedisInstance(opts)) {
      this.checkBlockingOptions(overrideMessage, opts);

      this.opts = {
        port: 6379,
        host: '127.0.0.1',
        retryStrategy: function (times: number) {
          return Math.max(Math.min(Math.exp(times), 20000), 1000);
        },
        ...opts,
      };

      if (this.blocking) {
        this.opts.maxRetriesPerRequest = null;
      }
    } else {
      this._client = opts;

      // Test if the redis instance is using keyPrefix
      // and if so, throw an error.
      if (this._client.options.keyPrefix) {
        throw new Error(
          'BullMQ: ioredis does not support ioredis prefixes, use the prefix option instead.',
        );
      }

      if (isRedisCluster(this._client)) {
        this.opts = this._client.options.redisOptions;
      } else {
        this.opts = this._client.options;
      }

      this.checkBlockingOptions(deprecationMessage, this.opts, true);
    }

    this.skipVersionCheck =
      skipVersionCheck || !!(this.opts && this.opts.skipVersionCheck);
    this.handleClientError = (err: Error): void => {
      this.emit('error', err);
    };

    this.handleClientClose = (): void => {
      this.emit('close');
    };

    this.handleClientReady = (): void => {
      this.emit('ready');
    };

    this.initializing = this.init();
    this.initializing.catch(err => this.emit('error', err));
  }

  private checkBlockingOptions(
    msg: string,
    options?: RedisOptions,
    throwError = false,
  ) {
    if (this.blocking && options && options.maxRetriesPerRequest) {
      if (throwError) {
        throw new Error(msg);
      } else {
        console.error(msg);
      }
    }
  }

  /**
   * Waits for a redis client to be ready.
   * @param redis - client
   */
  static async waitUntilReady(client: RedisClient): Promise<void> {
    if (client.status === 'ready') {
      return;
    }

    if (client.status === 'wait') {
      return client.connect();
    }

    if (client.status === 'end') {
      throw new Error(CONNECTION_CLOSED_ERROR_MSG);
    }

    let handleReady: () => void;
    let handleEnd: () => void;
    let handleError: (e: Error) => void;
    try {
      await new Promise<void>((resolve, reject) => {
        let lastError: Error;

        handleError = (err: Error) => {
          lastError = err;
        };

        handleReady = () => {
          resolve();
        };

        handleEnd = () => {
          reject(lastError || new Error(CONNECTION_CLOSED_ERROR_MSG));
        };

        increaseMaxListeners(client, 3);

        client.once('ready', handleReady);
        client.on('end', handleEnd);
        client.once('error', handleError);
      });
    } finally {
      client.removeListener('end', handleEnd);
      client.removeListener('error', handleError);
      client.removeListener('ready', handleReady);

      decreaseMaxListeners(client, 3);
    }
  }

  get client(): Promise<RedisClient> {
    return this.initializing;
  }

  protected loadCommands(providedScripts?: Record<string, RawCommand>): void {
    const finalScripts =
      providedScripts || (scripts as Record<string, RawCommand>);
    for (const property in finalScripts as Record<string, RawCommand>) {
      // Only define the command if not already defined
      if (!(<any>this._client)[finalScripts[property].name]) {
        (<any>this._client).defineCommand(finalScripts[property].name, {
          numberOfKeys: finalScripts[property].keys,
          lua: finalScripts[property].content,
        });
      }
    }
  }

  private async init() {
    if (!this._client) {
      this._client = new IORedis(this.opts);
    }

    increaseMaxListeners(this._client, 3);

    this._client.on('error', this.handleClientError);
    // ioredis treats connection errors as a different event ('close')
    this._client.on('close', this.handleClientClose);

    this._client.on('ready', this.handleClientReady);

    await RedisConnection.waitUntilReady(this._client);

    this.loadCommands();

    this.version = await this.getRedisVersion();
    if (this.skipVersionCheck !== true && !this.closing) {
      if (
        isRedisVersionLowerThan(this.version, RedisConnection.minimumVersion)
      ) {
        throw new Error(
          `Redis version needs to be greater or equal than ${RedisConnection.minimumVersion} Current: ${this.version}`,
        );
      }

      if (
        isRedisVersionLowerThan(
          this.version,
          RedisConnection.recommendedMinimumVersion,
        )
      ) {
        console.warn(
          `It is highly recommended to use a minimum Redis version of ${RedisConnection.recommendedMinimumVersion}
           Current: ${this.version}`,
        );
      }
    }

    this.capabilities = {
      canDoubleTimeout: !isRedisVersionLowerThan(this.version, '6.0.0'),
    };

    this.status = 'ready';

    return this._client;
  }

  async disconnect(wait = true): Promise<void> {
    const client = await this.client;
    if (client.status !== 'end') {
      let _resolve, _reject;

      if (!wait) {
        return client.disconnect();
      }

      const disconnecting = new Promise<void>((resolve, reject) => {
        increaseMaxListeners(client, 2);

        client.once('end', resolve);
        client.once('error', reject);
        _resolve = resolve;
        _reject = reject;
      });

      client.disconnect();

      try {
        await disconnecting;
      } finally {
        decreaseMaxListeners(client, 2);

        client.removeListener('end', _resolve);
        client.removeListener('error', _reject);
      }
    }
  }

  async reconnect(): Promise<void> {
    const client = await this.client;
    return client.connect();
  }

  async close(): Promise<void> {
    if (!this.closing) {
      const status = this.status;
      this.status = 'closing';
      this.closing = true;
      try {
        if (status === 'ready') {
          // Not sure if we need to wait for this
          await this.initializing;
          if (!this.shared) {
            await this._client.quit();
          }
        }
      } catch (error) {
        if (isNotConnectionError(error as Error)) {
          throw error;
        }
      } finally {
        this._client.off('error', this.handleClientError);
        this._client.off('close', this.handleClientClose);
        this._client.off('ready', this.handleClientReady);

        decreaseMaxListeners(this._client, 3);

        this.removeAllListeners();
        this.status = 'closed';
      }
    }
  }

  private async getRedisVersion() {
    const doc = await this._client.info();
    const redisPrefix = 'redis_version:';
    const maxMemoryPolicyPrefix = 'maxmemory_policy:';
    const lines = doc.split('\r\n');
    let redisVersion;

    for (let i = 0; i < lines.length; i++) {
      if (lines[i].indexOf(maxMemoryPolicyPrefix) === 0) {
        const maxMemoryPolicy = lines[i].substr(maxMemoryPolicyPrefix.length);
        if (maxMemoryPolicy !== 'noeviction') {
          console.warn(
            `IMPORTANT! Eviction policy is ${maxMemoryPolicy}. It should be "noeviction"`,
          );
        }
      }

      if (lines[i].indexOf(redisPrefix) === 0) {
        redisVersion = lines[i].substr(redisPrefix.length);
      }
    }

    return redisVersion;
  }

  get redisVersion(): string {
    return this.version;
  }
}

1	import { EventEmitter } from 'events';	1✔
2	import { default as IORedis } from 'ioredis';	1✔
3	// eslint-disable-next-line @typescript-eslint/ban-ts-comment
4	// @ts-ignore
5	import { CONNECTION_CLOSED_ERROR_MSG } from 'ioredis/built/utils';	1✔
6	import { ConnectionOptions, RedisOptions, RedisClient } from '../interfaces';
7	import {	1✔
8	decreaseMaxListeners,
9	increaseMaxListeners,
10	isNotConnectionError,
11	isRedisCluster,
12	isRedisInstance,
13	isRedisVersionLowerThan,
14	} from '../utils';
15	import * as scripts from '../scripts';	1✔
16
17	const overrideMessage = [	1✔
18	'BullMQ: WARNING! Your redis options maxRetriesPerRequest must be null',
19	'and will be overridden by BullMQ.',
20	].join(' ');
21
22	const deprecationMessage =
23	'BullMQ: Your redis options maxRetriesPerRequest must be null.';	1✔
24
25	interface RedisCapabilities {
26	canDoubleTimeout: boolean;
27	}
28
29	export interface RawCommand {
30	content: string;
31	name: string;
32	keys: number;
33	}
34
35	export class RedisConnection extends EventEmitter {	1✔
36	static minimumVersion = '5.0.0';	1✔
37	static recommendedMinimumVersion = '6.2.0';	1✔
38
39	closing: boolean;
40	capabilities: RedisCapabilities = {	1,712✔
41	canDoubleTimeout: false,
42	};
43
44	status: 'initializing' \| 'ready' \| 'closing' \| 'closed' = 'initializing';	1,712✔
45
46	protected _client: RedisClient;
47
48	private readonly opts: RedisOptions;
49	private readonly initializing: Promise<RedisClient>;
50
51	private version: string;
52	private skipVersionCheck: boolean;
53	private handleClientError: (e: Error) => void;
54	private handleClientClose: () => void;
55	private handleClientReady: () => void;
56
57	constructor(
58	opts?: ConnectionOptions,
59	private readonly shared: boolean = false,	1,712✔
60	private readonly blocking = true,	1,712✔
61	skipVersionCheck = false,	1,712✔
62	) {
63	super();	1,712✔
64
65	if (!isRedisInstance(opts)) {	1,712✔
66	this.checkBlockingOptions(overrideMessage, opts);	7✔
67
68	this.opts = {	7✔
69	port: 6379,
70	host: '127.0.0.1',
71	retryStrategy: function (times: number) {
72	return Math.max(Math.min(Math.exp(times), 20000), 1000);	×
73	},
74	...opts,
75	};
76
77	if (this.blocking) {	7✔
78	this.opts.maxRetriesPerRequest = null;	3✔
79	}
80	} else {
81	this._client = opts;	1,705✔
82
83	// Test if the redis instance is using keyPrefix
84	// and if so, throw an error.
85	if (this._client.options.keyPrefix) {	1,705✔
86	throw new Error(	2✔
87	'BullMQ: ioredis does not support ioredis prefixes, use the prefix option instead.',
88	);
89	}
90
91	if (isRedisCluster(this._client)) {	1,703✔
92	this.opts = this._client.options.redisOptions;	1✔
93	} else {
94	this.opts = this._client.options;	1,702✔
95	}
96
97	this.checkBlockingOptions(deprecationMessage, this.opts, true);	1,703✔
98	}
99
100	this.skipVersionCheck =	1,710✔
101	skipVersionCheck \|\| !!(this.opts && this.opts.skipVersionCheck);	6,839✔
102	this.handleClientError = (err: Error): void => {	1,710✔
103	this.emit('error', err);	159✔
104	};
105
106	this.handleClientClose = (): void => {	1,710✔
107	this.emit('close');	852✔
108	};
109
110	this.handleClientReady = (): void => {	1,710✔
111	this.emit('ready');	702✔
112	};
113
114	this.initializing = this.init();	1,710✔
115	this.initializing.catch(err => this.emit('error', err));	1,710✔
116	}
117
118	private checkBlockingOptions(
119	msg: string,
120	options?: RedisOptions,
121	throwError = false,	7✔
122	) {
123	if (this.blocking && options && options.maxRetriesPerRequest) {	1,710!
124	if (throwError) {	×
125	throw new Error(msg);	×
126	} else {
127	console.error(msg);	×
128	}
129	}
130	}
131
132	/**
133	* Waits for a redis client to be ready.
134	* @param redis - client
135	*/
136	static async waitUntilReady(client: RedisClient): Promise<void> {
137	if (client.status === 'ready') {	1,712✔
138	return;	1,008✔
139	}
140
141	if (client.status === 'wait') {	704!
142	return client.connect();	×
143	}
144
145	if (client.status === 'end') {	704!
146	throw new Error(CONNECTION_CLOSED_ERROR_MSG);	×
147	}
148
149	let handleReady: () => void;
150	let handleEnd: () => void;
151	let handleError: (e: Error) => void;
152	try {	704✔
153	await new Promise<void>((resolve, reject) => {	704✔
154	let lastError: Error;
155
156	handleError = (err: Error) => {	704✔
157	lastError = err;	6✔
158	};
159
160	handleReady = () => {	704✔
161	resolve();	697✔
162	};
163
164	handleEnd = () => {	704✔
165	reject(lastError \|\| new Error(CONNECTION_CLOSED_ERROR_MSG));	4!
166	};
167
168	increaseMaxListeners(client, 3);	704✔
169
170	client.once('ready', handleReady);	704✔
171	client.on('end', handleEnd);	704✔
172	client.once('error', handleError);	704✔
173	});
174	} finally {
175	client.removeListener('end', handleEnd);	701✔
176	client.removeListener('error', handleError);	701✔
177	client.removeListener('ready', handleReady);	701✔
178
179	decreaseMaxListeners(client, 3);	701✔
180	}
181	}
182
183	get client(): Promise<RedisClient> {
184	return this.initializing;	33,598✔
185	}
186
187	protected loadCommands(providedScripts?: Record<string, RawCommand>): void {
188	const finalScripts =
189	providedScripts \|\| (scripts as Record<string, RawCommand>);	1,703✔
190	for (const property in finalScripts as Record<string, RawCommand>) {	1,703✔
191	// Only define the command if not already defined
192	if (!(<any>this._client)[finalScripts[property].name]) {	57,902✔
193	(<any>this._client).defineCommand(finalScripts[property].name, {	23,528✔
194	numberOfKeys: finalScripts[property].keys,
195	lua: finalScripts[property].content,
196	});
197	}
198	}
199	}
200
201	private async init() {
202	if (!this._client) {	1,710✔
203	this._client = new IORedis(this.opts);	7✔
204	}
205
206	increaseMaxListeners(this._client, 3);	1,710✔
207
208	this._client.on('error', this.handleClientError);	1,710✔
209	// ioredis treats connection errors as a different event ('close')
210	this._client.on('close', this.handleClientClose);	1,710✔
211
212	this._client.on('ready', this.handleClientReady);	1,710✔
213
214	await RedisConnection.waitUntilReady(this._client);	1,710✔
215
216	this.loadCommands();	1,703✔
217
218	this.version = await this.getRedisVersion();	1,703✔
219	if (this.skipVersionCheck !== true && !this.closing) {	1,703✔
220	if (	1,696!
221	isRedisVersionLowerThan(this.version, RedisConnection.minimumVersion)
222	) {
223	throw new Error(	×
224	`Redis version needs to be greater or equal than ${RedisConnection.minimumVersion} Current: ${this.version}`,
225	);
226	}
227
228	if (	1,696!
229	isRedisVersionLowerThan(
230	this.version,
231	RedisConnection.recommendedMinimumVersion,
232	)
233	) {
234	console.warn(	×
235	`It is highly recommended to use a minimum Redis version of ${RedisConnection.recommendedMinimumVersion}
236	Current: ${this.version}`,
237	);
238	}
239	}
240
241	this.capabilities = {	1,703✔
242	canDoubleTimeout: !isRedisVersionLowerThan(this.version, '6.0.0'),
243	};
244
245	this.status = 'ready';	1,703✔
246
247	return this._client;	1,703✔
248	}
249
250	async disconnect(wait = true): Promise<void> {	291✔
251	const client = await this.client;	407✔
252	if (client.status !== 'end') {	407!
253	let _resolve, _reject;
254
255	if (!wait) {	407✔
256	return client.disconnect();	114✔
257	}
258
259	const disconnecting = new Promise<void>((resolve, reject) => {	293✔
260	increaseMaxListeners(client, 2);	293✔
261
262	client.once('end', resolve);	293✔
263	client.once('error', reject);	293✔
264	_resolve = resolve;	293✔
265	_reject = reject;	293✔
266	});
267
268	client.disconnect();	293✔
269
270	try {	293✔
271	await disconnecting;	293✔
272	} finally {
273	decreaseMaxListeners(client, 2);	293✔
274
275	client.removeListener('end', _resolve);	293✔
276	client.removeListener('error', _reject);	293✔
277	}
278	}
279	}
280
281	async reconnect(): Promise<void> {
282	const client = await this.client;	2✔
283	return client.connect();	2✔
284	}
285
286	async close(): Promise<void> {
287	if (!this.closing) {	1,004!
288	const status = this.status;	1,004✔
289	this.status = 'closing';	1,004✔
290	this.closing = true;	1,004✔
291	try {	1,004✔
292	if (status === 'ready') {	1,004✔
293	// Not sure if we need to wait for this
294	await this.initializing;	994✔
295	if (!this.shared) {	994!
NEW 296	await this._client.quit();	×
297	}
298	}
299	} catch (error) {
UNCOV 300	if (isNotConnectionError(error as Error)) {	×
301	throw error;	×
302	}
303	} finally {
304	this._client.off('error', this.handleClientError);	1,004✔
305	this._client.off('close', this.handleClientClose);	1,004✔
306	this._client.off('ready', this.handleClientReady);	1,004✔
307
308	decreaseMaxListeners(this._client, 3);	1,004✔
309
310	this.removeAllListeners();	1,004✔
311	this.status = 'closed';	1,004✔
312	}
313	}
314	}
315
316	private async getRedisVersion() {
317	const doc = await this._client.info();	1,703✔
318	const redisPrefix = 'redis_version:';	1,703✔
319	const maxMemoryPolicyPrefix = 'maxmemory_policy:';	1,703✔
320	const lines = doc.split('\r\n');	1,703✔
321	let redisVersion;
322
323	for (let i = 0; i < lines.length; i++) {	1,703✔
324	if (lines[i].indexOf(maxMemoryPolicyPrefix) === 0) {	367,822✔
325	const maxMemoryPolicy = lines[i].substr(maxMemoryPolicyPrefix.length);	1,703✔
326	if (maxMemoryPolicy !== 'noeviction') {	1,703!
327	console.warn(	×
328	`IMPORTANT! Eviction policy is ${maxMemoryPolicy}. It should be "noeviction"`,
329	);
330	}
331	}
332
333	if (lines[i].indexOf(redisPrefix) === 0) {	367,822✔
334	redisVersion = lines[i].substr(redisPrefix.length);	1,703✔
335	}
336	}
337
338	return redisVersion;	1,703✔
339	}
340
341	get redisVersion(): string {
342	return this.version;	345✔
343	}
344	}

taskforcesh / bullmq / 7298903038

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous