11571942445

Committed 29 Oct 2024 10:27AM UTC coverage: 88.018% (-0.8%) from 88.819%

Build # 11571942445

Build Type

Pull #2721

github

Committed by

web-flow

Commit Message

Merge 210d36e04 into 0547f7a80

Pull Request Pull Request #2721: feat(queue): observability for queue

Run Details

1381 of 1669 branches covered (82.74%)

Branch coverage included in aggregate %.

281 of 310 new or added lines in 6 files covered. (90.65%)

66 existing lines in 5 files now uncovered.

2527 of 2771 relevant lines covered (91.19%)

4013.96 hits per line

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

88.69

/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 { version } from '../version';
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;
  canBlockFor1Ms: 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,
    canBlockFor1Ms: true,
  };

  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 = () => {
          if (client.status !== 'end') {
            reject(lastError || new Error(CONNECTION_CLOSED_ERROR_MSG));
          } else {
            if (lastError) {
              reject(lastError);
            } else {
              // when custon 'end' status is set we already closed
              resolve();
            }
          }
        };

        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(
    version: string,
    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
      const commandName = `${finalScripts[property].name}:${version}`;
      if (!(<any>this._client)[commandName]) {
        (<any>this._client).defineCommand(commandName, {
          numberOfKeys: finalScripts[property].keys,
          lua: finalScripts[property].content,
        });
      }
    }
  }

  private async init() {
    if (!this._client) {
      const { url, ...rest } = this.opts;
      this._client = url ? new IORedis(url, rest) : new IORedis(rest);
    }

    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(version);

    if (this._client['status'] !== 'end') {
      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'),
        canBlockFor1Ms: !isRedisVersionLowerThan(this.version, '7.0.8'),
      };

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

taskforcesh / bullmq / 11571942445

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